Merge pull request #24 from wide-vsix/dev

LogOutputHook Feature

config.yaml

@@ -25,7 +25,22 @@ outputs:
     # nfdump -r /tmp/netflow/nfcapd.202207101030 -o extended
 - log:
     file: /tmp/flow.log
-
+    # hooks:
+    # - name: hostname addition
+    #   command: /usr/bin/hook_command_example_hostname.sh
+    # - name: shell to resolve hostname
+    #   shell: |
+    #     #!/bin/sh
+    #     echo `cat` | jq --arg hostname $(hostname) '. + {hostname: $hostname}'
+    # - name: shell to resolve ifname from ifindex
+    #   shell: |
+    #     #!/bin/sh
+    #     IN=$(cat)
+    #     I_IDX=$(echo $IN | jq .ingressIfindex -r)
+    #     E_IDX=$(echo $IN | jq .egressIfindex -r )
+    #     I_NAME=$(ip -n ns0 -j link | jq --argjson idx $I_IDX '.[] | select(.ifindex == $idx) | .ifname' -r)
+    #     E_NAME=$(ip -n ns0 -j link | jq --argjson idx $E_IDX '.[] | select(.ifindex == $idx) | .ifname' -r)
+    #     echo $IN | jq --arg i_name $I_NAME --arg e_name $E_NAME '. + {ingressIfname: $i_name, egressIfname: $e_name}'
 templates:
 - id: 1001
   template:

examples/agent_output_hook/config.yaml

@@ -0,0 +1,23 @@
+maxIpfixMessageLen: 100
+timerFinishedDrainSeconds: 1
+timerForceDrainSeconds: 30
+timerTemplateFlushSeconds: 60
+outputs:
+- log:
+    file: /tmp/flowlog.json
+    hooks:
+    - name: hostname addition
+      command: ./misc/hook_command_example_dummy.sh
+    - name: shell to resolve hostname
+      shell: |
+        #!/bin/sh
+        echo `cat` | jq --arg hostname $(hostname) '. + {hostname: $hostname}'
+    - name: shell to resolve ifname from ifindex
+      shell: |
+        #!/bin/sh
+        IN=$(cat)
+        I_IDX=$(echo $IN | jq .ingressIfindex -r)
+        E_IDX=$(echo $IN | jq .egressIfindex -r )
+        I_NAME=$(ip -n ns0 -j link | jq --argjson idx $I_IDX '.[] | select(.ifindex == $idx) | .ifname' -r)
+        E_NAME=$(ip -n ns0 -j link | jq --argjson idx $E_IDX '.[] | select(.ifindex == $idx) | .ifname' -r)
+        echo $IN | jq --arg i_name $I_NAME --arg e_name $E_NAME '. + {ingressIfname: $i_name, egressIfname: $e_name}'

examples/agent_output_hook/flowlog.json

@@ -0,0 +1,46 @@
+{
+  "level": "info",
+  "ts": 1667266821.7434478,
+  "caller": "flowctl/ipfix.go:430",
+  "msg": "flowlog",
+  "bytes": 594,
+  "finished": 1,
+  "foo": "bar",
+  "pkts": 6,
+  "proto": 6,
+  "start": 12168400776634952,
+  "action": 0,
+  "ingressIfindex": 3,
+  "dport": 32816,
+  "egressIfindex": 2,
+  "hostname": "slankdev",
+  "src": "10.2.0.2",
+  "ingressIfname": "eth2",
+  "egressIfname": "eth1",
+  "dst": "10.1.0.2",
+  "sport": 8080,
+  "end": 12168400778056844
+}
+{
+  "level": "info",
+  "ts": 1667266821.9629853,
+  "caller": "flowctl/ipfix.go:430",
+  "msg": "flowlog",
+  "dport": 8080,
+  "dst": "10.2.0.2",
+  "proto": 6,
+  "src": "10.1.0.2",
+  "finished": 1,
+  "sport": 32816,
+  "start": 12168400776608754,
+  "ingressIfname": "eth1",
+  "egressIfindex": 3,
+  "foo": "bar",
+  "egressIfname": "eth2",
+  "pkts": 6,
+  "action": 0,
+  "bytes": 481,
+  "end": 12168400778037424,
+  "hostname": "slankdev",
+  "ingressIfindex": 2
+}

misc/hook_command_example_dummy.sh

@@ -0,0 +1,18 @@
+#!/bin/sh
+# IN:
+# {
+#   "src": "10.1.0.1",
+#   "dst": "10.2.0.1",
+#   "pkts": 10,
+#   "bytes": 1000
+# }
+#
+# OUT:
+# {
+#   "src": "10.1.0.1",
+#   "dst": "10.2.0.1",
+#   "pkts": 10,
+#   "bytes": 1000,
+#   "foo": "bar"
+# }
+echo `cat` | jq '. + {foo: "bar"}'

misc/hook_command_example_hostname.sh

@@ -0,0 +1,18 @@
+#!/bin/sh
+# IN:
+# {
+#   "src": "10.1.0.1",
+#   "dst": "10.2.0.1",
+#   "pkts": 10,
+#   "bytes": 1000
+# }
+#
+# OUT:
+# {
+#   "src": "10.1.0.1",
+#   "dst": "10.2.0.1",
+#   "pkts": 10,
+#   "bytes": 1000,
+#   "hostname": "machine1"
+# }
+echo `cat` | jq --arg hostname $(hostname) '. + {hostname: $hostname}'

misc/hook_command_example_ifname.sh

@@ -0,0 +1,24 @@
+#!/bin/sh
+# IN:
+# {
+#   "ingressIfindex": 1,
+#   "egressIfindex": 2,
+#   "pkts": 10,
+#   "bytes": 1000
+# }
+#
+# OUT:
+# {
+#   "ingressIfindex": 1,
+#   "egressIfindex": 2,
+#   "ingressIfname": 1,
+#   "egressIfname": 2,
+#   "pkts": 10,
+#   "bytes": 1000
+# }
+IN=$(cat)
+I_IDX=$(echo $IN | jq .ingressIfindex -r)
+E_IDX=$(echo $IN | jq .egressIfindex -r )
+I_NAME=$(ip -n ns0 -j link | jq --argjson idx $I_IDX '.[] | select(.ifindex == $idx) | .ifname' -r)
+E_NAME=$(ip -n ns0 -j link | jq --argjson idx $E_IDX '.[] | select(.ifindex == $idx) | .ifname' -r)
+echo $IN | jq --arg i_name $I_NAME --arg e_name $E_NAME '. + {ingressIfname: $i_name, egressIfname: $e_name}'

pkg/ebpfmap/types.go

@@ -308,20 +308,34 @@ func ToIpfixFlowFile(ebflows []Flow) (*ipfix.FlowFile, error) {
 	return flowFile, nil
 }
 
-func (f Flow) ToZap() []interface{} {
-	return []interface{}{
-		"src", util.ConvertUint32ToIP(f.Key.Saddr).String(),
-		"dst", util.ConvertUint32ToIP(f.Key.Daddr).String(),
-		"proto", f.Key.Proto,
-		"sport", f.Key.Sport,
-		"dport", f.Key.Dport,
-		"ingressIfindex", f.Key.IngressIfindex,
-		"egressIfindex", f.Key.EgressIfindex,
-		"pkts", f.Val.FlowPkts,
-		"bytes", f.Val.FlowBytes,
-		"action", f.Key.Mark,
-		"start", f.Val.FlowStartMilliSecond,
-		"end", f.Val.FlowEndMilliSecond,
-		"finished", f.Val.Finished,
+func (f Flow) ToZap(o ipfix.OutputLog) ([]interface{}, error) {
+	m := map[string]interface{}{
+		"src":            util.ConvertUint32ToIP(f.Key.Saddr).String(),
+		"dst":            util.ConvertUint32ToIP(f.Key.Daddr).String(),
+		"proto":          f.Key.Proto,
+		"sport":          f.Key.Sport,
+		"dport":          f.Key.Dport,
+		"ingressIfindex": f.Key.IngressIfindex,
+		"egressIfindex":  f.Key.EgressIfindex,
+		"pkts":           f.Val.FlowPkts,
+		"bytes":          f.Val.FlowBytes,
+		"action":         f.Key.Mark,
+		"start":          f.Val.FlowStartMilliSecond,
+		"end":            f.Val.FlowEndMilliSecond,
+		"finished":       f.Val.Finished,
 	}
+
+	for _, h := range o.Hooks {
+		var err error
+		m, err = h.Execute(m)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	ret := []interface{}{}
+	for key, val := range m {
+		ret = append(ret, key, val)
+	}
+	return ret, nil
 }

pkg/flowctl/ipfix.go

@@ -194,7 +194,7 @@ func fnIpfixDump(cmd *cobra.Command, args []string) error {
 				return fmt.Errorf("invalid config")
 			}
 			if o.Log != nil {
-				if err := FlowOutputLog(ebpfFlows, o.Log.File); err != nil {
+				if err := FlowOutputLog(ebpfFlows, o.Log.File, *o.Log); err != nil {
 					return err
 				}
 			}
@@ -345,7 +345,7 @@ func flushCachesFinished(config ipfix.Config) error {
 			return fmt.Errorf("invalid config")
 		}
 		if o.Log != nil {
-			if err := FlowOutputLog(ebpfFlows, o.Log.File); err != nil {
+			if err := FlowOutputLog(ebpfFlows, o.Log.File, *o.Log); err != nil {
 				return err
 			}
 		}
@@ -385,7 +385,7 @@ func flushCaches(config ipfix.Config) error {
 			return fmt.Errorf("invalid config")
 		}
 		if o.Log != nil {
-			if err := FlowOutputLog(ebpfFlows, o.Log.File); err != nil {
+			if err := FlowOutputLog(ebpfFlows, o.Log.File, *o.Log); err != nil {
 				return err
 			}
 		}
@@ -415,7 +415,7 @@ func flushCaches(config ipfix.Config) error {
 	return nil
 }
 
-func FlowOutputLog(flows []ebpfmap.Flow, out string) error {
+func FlowOutputLog(flows []ebpfmap.Flow, out string, o ipfix.OutputLog) error {
 	cfg := zap.NewProductionConfig()
 	cfg.OutputPaths = []string{
 		out,
@@ -427,7 +427,11 @@ func FlowOutputLog(flows []ebpfmap.Flow, out string) error {
 	log := zapr.NewLogger(zapLog)
 
 	for _, flow := range flows {
-		log.Info("flowlog", flow.ToZap()...)
+		args, err := flow.ToZap(o)
+		if err != nil {
+			return err
+		}
+		log.Info("flowlog", args...)
 	}
 	return nil
 }

pkg/hook/command.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2022 Hiroki Shirokura.
+Copyright 2022 LINE Corporation.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package hook
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"os/exec"
+)
+
+type Command string
+
+var _ Hook = (*Command)(nil)
+
+func (c Command) Execute(in map[string]interface{}) (map[string]interface{}, error) {
+	// Prepare input/output
+	stdoutbuf := bytes.Buffer{}
+	stderrbuf := bytes.Buffer{}
+	stdinbytes, err := json.Marshal(in)
+	if err != nil {
+		return nil, err
+	}
+
+	// Execute child process
+	cmd := exec.Command(string(c))
+	cmd.Stdout = &stdoutbuf
+	cmd.Stderr = &stderrbuf
+	cmd.Stdin = bytes.NewBuffer(stdinbytes)
+	if err := cmd.Run(); err != nil {
+		return nil, fmt.Errorf("child process is failed: err=%v stderr=%s",
+			err, stderrbuf.String())
+	}
+
+	// Convert back to map data from json-bytes
+	out := map[string]interface{}{}
+	if err := json.Unmarshal(stdoutbuf.Bytes(), &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}

pkg/hook/interface.go

@@ -0,0 +1,22 @@
+/*
+Copyright 2022 Hiroki Shirokura.
+Copyright 2022 LINE Corporation.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package hook
+
+type Hook interface {
+	Execute(in map[string]interface{}) (map[string]interface{}, error)
+}