[HOLD] Add support for payload via event transport

cli/cmd/attach.go

@@ -69,6 +69,8 @@ be set to sockets with unix:///var/run/mysock, tcp://hostname:port, udp://hostna
 			helpErrAndExit(cmd, "Cannot specify --verbosity and --userconfig")
 		} else if rc.Payloads && rc.UserConfig != "" {
 			helpErrAndExit(cmd, "Cannot specify --payloads and --userconfig")
+		} else if cmd.Flags().Lookup("payloadsdest").Changed && rc.UserConfig != "" {
+			helpErrAndExit(cmd, "Cannot specify --payloadsdest and --userconfig")
 		} else if rc.Loglevel != "" && rc.UserConfig != "" {
 			helpErrAndExit(cmd, "Cannot specify --loglevel and --userconfig")
 		} else if rc.Backtrace && rc.UserConfig != "" {

cli/cmd/rules.go

@@ -61,7 +61,8 @@ var rulesCmd = &cobra.Command{
 		if addProc == "" && remProc == "" {
 			if rc.CriblDest != "" || rc.MetricsDest != "" || rc.EventsDest != "" || rc.UserConfig != "" ||
 				cmd.Flags().Lookup("metricformat").Changed || rc.NoBreaker || rc.AuthToken != "" || rc.Payloads ||
-				rc.Backtrace || rc.Loglevel != "" || rc.Coredump || cmd.Flags().Lookup("verbosity").Changed {
+				cmd.Flags().Lookup("payloadsdest").Changed || rc.Backtrace || rc.Loglevel != "" || rc.Coredump ||
+				cmd.Flags().Lookup("verbosity").Changed {
 				helpErrAndExit(cmd, "The rules command without --add or --remove options, only supports the --json flag")
 			}
 		}
@@ -86,6 +87,8 @@ var rulesCmd = &cobra.Command{
 			helpErrAndExit(cmd, "Cannot specify --verbosity and --userconfig")
 		} else if rc.Payloads && rc.UserConfig != "" {
 			helpErrAndExit(cmd, "Cannot specify --payloads and --userconfig")
+		} else if cmd.Flags().Lookup("payloadsdest").Changed && rc.UserConfig != "" {
+			helpErrAndExit(cmd, "Cannot specify --payloadsdest and --userconfig")
 		} else if rc.Loglevel != "" && rc.UserConfig != "" {
 			helpErrAndExit(cmd, "Cannot specify --loglevel and --userconfig")
 		} else if rc.Backtrace && rc.UserConfig != "" {

cli/cmd/run.go

@@ -63,6 +63,8 @@ The --*dest flags accept file names like /tmp/scope.log; URLs like file:///tmp/s
 			helpErrAndExit(cmd, "Cannot specify --verbosity and --userconfig")
 		} else if rc.Payloads && rc.UserConfig != "" {
 			helpErrAndExit(cmd, "Cannot specify --payloads and --userconfig")
+		} else if cmd.Flags().Lookup("payloadsdest").Changed && rc.UserConfig != "" {
+			helpErrAndExit(cmd, "Cannot specify --payloadsdest and --userconfig")
 		} else if rc.Loglevel != "" && rc.UserConfig != "" {
 			helpErrAndExit(cmd, "Cannot specify --loglevel and --userconfig")
 		} else if rc.Backtrace && rc.UserConfig != "" {

cli/cmd/util.go

@@ -59,6 +59,7 @@ func metricAndEventDestFlags(cmd *cobra.Command, rc *run.Config) {
 func runCmdFlags(cmd *cobra.Command, rc *run.Config) {
 	cmd.Flags().IntVarP(&rc.Verbosity, "verbosity", "v", 4, "Set scope metric verbosity")
 	cmd.Flags().BoolVarP(&rc.Payloads, "payloads", "p", false, "Capture payloads of network transactions")
+	cmd.Flags().StringVarP(&rc.PayloadsDest, "payloadsdest", "", "dir", "Set destination for payloads (dir|event)")
 	cmd.Flags().StringVar(&rc.Loglevel, "loglevel", "", "Set scope library log level (debug, warning, info, error, none)")
 	cmd.Flags().StringVarP(&rc.LibraryPath, "librarypath", "l", "", "Set path for dynamic libraries")
 	cmd.Flags().StringVarP(&rc.UserConfig, "userconfig", "u", "", "Scope an application with a user specified config file; overrides all other settings.")

cli/libscope/config.go

@@ -67,6 +67,7 @@ type ScopeEventConfig struct {
 // ScopePayloadConfig represents how to capture payloads
 type ScopePayloadConfig struct {
 	Enable BoolString `mapstructure:"enable" json:"enable" yaml:"enable"`
+	Type   string     `mapstructure:"type" json:"type" yaml:"type"`
 	Dir    string     `mapstructure:"dir" json:"dir" yaml:"dir"`
 }
 

cli/run/run.go

@@ -15,6 +15,7 @@ type Config struct {
 	WorkDir       string
 	Verbosity     int
 	Payloads      bool
+	PayloadsDest  string
 	MetricsDest   string
 	EventsDest    string
 	MetricsFormat string

cli/run/scopeconfig.go

@@ -105,6 +105,10 @@ func (c *Config) SetDefault() error {
 				// },
 			},
 		},
+		Payload: libscope.ScopePayloadConfig{
+			Enable: "false",
+			Type:   "dir",
+			Dir:    filepath.Join(c.WorkDir, "payloads")},
 		Libscope: libscope.ScopeLibscopeConfig{
 			SummaryPeriod: 10,
 			CommandDir:    filepath.Join(c.WorkDir, "cmd"),
@@ -191,10 +195,19 @@ func (c *Config) configFromRunOpts() error {
 	if c.Payloads {
 		c.sc.Payload = libscope.ScopePayloadConfig{
 			Enable: "true",
+			Type:   "dir",
 			Dir:    filepath.Join(c.WorkDir, "payloads"),
 		}
 	}
 
+	if c.PayloadsDest != "" {
+		if c.PayloadsDest != "dir" && c.PayloadsDest != "event" {
+			return fmt.Errorf("invalid payload destination %s", c.PayloadsDest)
+		}
+		c.sc.Payload.Type = c.PayloadsDest
+		// TODO should we empty c.sc.Payload.Dir as well ?
+	}
+
 	if c.MetricsFormat != "" {
 		if c.MetricsFormat != "ndjson" && c.MetricsFormat != "statsd" {
 			return fmt.Errorf("invalid metrics format %s", c.MetricsFormat)

cli/run/scopeconfig_test.go

@@ -80,8 +80,19 @@ func TestConfigFromRunOpts(t *testing.T) {
 	err = c.configFromRunOpts()
 	assert.NoError(t, err)
 	assert.EqualValues(t, "true", c.sc.Payload.Enable)
+	assert.EqualValues(t, "dir", c.sc.Payload.Type)
 	assert.Equal(t, ".foo/payloads", c.sc.Payload.Dir)
 
+	c.PayloadsDest = "foo"
+	err = c.configFromRunOpts()
+	assert.Error(t, err)
+
+	c.PayloadsDest = "event"
+	err = c.configFromRunOpts()
+	assert.NoError(t, err)
+	assert.NoError(t, err)
+	assert.Equal(t, "event", c.sc.Payload.Type)
+
 	c.MetricsFormat = "foo"
 	err = c.configFromRunOpts()
 	assert.Error(t, err)

cli/run/setup_test.go

@@ -159,6 +159,10 @@ event:
     name: .*
     field: .*
     value: .*
+payload:
+  enable: false
+  type: dir
+  dir: PAYLOADSPATH
 libscope:
   configevent: true
   summaryperiod: 10
@@ -181,6 +185,7 @@ libscope:
 	expectedYaml = strings.Replace(expectedYaml, "VERBOSITY", strconv.Itoa(verbosity), 1)
 	expectedYaml = strings.Replace(expectedYaml, "METRICSPATH", filepath.Join(wd, "metrics.json"), 1)
 	expectedYaml = strings.Replace(expectedYaml, "EVENTSPATH", filepath.Join(wd, "events.json"), 1)
+	expectedYaml = strings.Replace(expectedYaml, "PAYLOADSPATH", filepath.Join(wd, "payloads"), 1)
 	expectedYaml = strings.Replace(expectedYaml, "CMDDIR", filepath.Join(wd, "cmd"), 1)
 	expectedYaml = strings.Replace(expectedYaml, "LIBSCOPELOGPATH", filepath.Join(wd, "libscope.log"), 1)
 	return expectedYaml

conf/scope.yml

@@ -604,6 +604,19 @@ payload:
   #
   enable: false
 
+  # Determine the payload type destination
+  #   Type:     string
+  #   Values:   "dir", "event"
+  #   Default:  "dir"
+  #   Override: $SCOPE_PAYLOAD_DEST
+  #
+  #
+  # This allows to specify the payload destination
+  # - "event" allows to send the payloads to same location as events
+  # - "dir" allows to use directory to store payload files
+  #
+  type: "dir"
+
   # Directory for payload files
   #   Type:     string
   #   Values:   (directory path)
@@ -612,6 +625,8 @@ payload:
   #
   # Consider using a performant filesystem to reduce I/O performance impacts.
   #
+  # Applies when dest is "dir".
+  #
   dir: '/tmp'
 
 # Setting up the AppScope library

docs/Help.md

@@ -243,6 +243,11 @@ Environment Variables:
         Default is /tmp
     SCOPE_PAYLOAD_ENABLE
         Flag that enables payload capture.  true,false  Default is false.
+    SCOPE_PAYLOAD_DEST
+        Specifies a payload destination. Possible values, are:
+        - 'dir' - payload files are captured in the directory
+        - 'event' - payload files are captured in same destination as event
+        Default is dir.
     SCOPE_PAYLOAD_DIR
         Specifies a directory where payload capture files can be written.
         Default is /tmp

docs/schemas/definitions/data.schema.json

@@ -448,6 +448,12 @@
       "type": "string",
       "enum": ["true", "false"]
     },
+    "payload_type": {
+      "title": "type",
+      "description": "Specifies the transport mechanism on which to emit the network payloads. See `scope.yml`.",
+      "type": "string",
+      "enum": ["dir", "event"]
+    },
     "pid": {
       "title": "pid",
       "description": "The process ID of the scoped app.",

docs/schemas/event_start_msg.schema.json

@@ -4,7 +4,7 @@
   "type": "object",
   "title": "AppScope Start message",
   "description": "Structure of the process-start message",
-  "examples": [{"format":"ndjson","info":{"process":{"libscopever":"v1.3.0","pid":35673,"ppid":3390,"gid":1000,"groupname":"test_user","uid":1000,"username":"test_user","hostname":"test_user","procname":"ls","cmd":"ls --color=auto","id":"test_user-ls-ls --color=auto","cgroup":"9:cpuset:/","machine_id":"a1e2ada5a5b1b273b4b5c0c2c1c4f5d1","uuid":"da845a9b-a55d-4c42-893d-08b54ee6e999"},"configuration":{"current":{"metric":{"enable":"true","transport":{"type":"udp","host":"127.0.0.1","port":"8125","tls":{"enable":"false","validateserver":"true","cacertpath":""}},"format":{"type":"statsd","statsdprefix":"","statsdmaxlen":512,"verbosity":4},"watch":[{"type":"fs"},{"type":"net"},{"type":"http"},{"type":"dns"},{"type":"process"},{"type":"statsd"}]},"libscope":{"log":{"level":"info","transport":{"type":"file","path":"/tmp/scope.log","buffering":"line"}},"snapshot":{"coredump":"false","backtrace":"false"},"configevent":"true","summaryperiod":10,"commanddir":"/tmp"},"event":{"enable":"true","transport":{"type":"tcp","host":"127.0.0.1","port":"9109","tls":{"enable":"false","validateserver":"true","cacertpath":""}},"format":{"type":"ndjson","maxeventpersec":10000,"enhancefs":"true"},"watch":[{"type":"file","name":"(\\/logs?\\/)|(\\.log$)|(\\.log[.\\d])","field":".*","value":".*"},{"type":"console","name":"(stdout)|(stderr)","field":".*","value":".*","allowbinary":"true"},{"type":"http","name":".*","field":".*","value":".*","headers":[]},{"type":"net","name":".*","field":".*","value":".*"},{"type":"fs","name":".*","field":".*","value":".*"},{"type":"dns","name":".*","field":".*","value":".*"}]},"payload":{"enable":"false","dir":"/tmp"},"tags":{},"protocol":[],"cribl":{"enable":"false","transport":{"type":"edge"},"authtoken":""}}},"environment":{}}}],
+  "examples": [{"format":"ndjson","info":{"process":{"libscopever":"v1.3.0","pid":35673,"ppid":3390,"gid":1000,"groupname":"test_user","uid":1000,"username":"test_user","hostname":"test_user","procname":"ls","cmd":"ls --color=auto","id":"test_user-ls-ls --color=auto","cgroup":"9:cpuset:/","machine_id":"a1e2ada5a5b1b273b4b5c0c2c1c4f5d1","uuid":"da845a9b-a55d-4c42-893d-08b54ee6e999"},"configuration":{"current":{"metric":{"enable":"true","transport":{"type":"udp","host":"127.0.0.1","port":"8125","tls":{"enable":"false","validateserver":"true","cacertpath":""}},"format":{"type":"statsd","statsdprefix":"","statsdmaxlen":512,"verbosity":4},"watch":[{"type":"fs"},{"type":"net"},{"type":"http"},{"type":"dns"},{"type":"process"},{"type":"statsd"}]},"libscope":{"log":{"level":"info","transport":{"type":"file","path":"/tmp/scope.log","buffering":"line"}},"snapshot":{"coredump":"false","backtrace":"false"},"configevent":"true","summaryperiod":10,"commanddir":"/tmp"},"event":{"enable":"true","transport":{"type":"tcp","host":"127.0.0.1","port":"9109","tls":{"enable":"false","validateserver":"true","cacertpath":""}},"format":{"type":"ndjson","maxeventpersec":10000,"enhancefs":"true"},"watch":[{"type":"file","name":"(\\/logs?\\/)|(\\.log$)|(\\.log[.\\d])","field":".*","value":".*"},{"type":"console","name":"(stdout)|(stderr)","field":".*","value":".*","allowbinary":"true"},{"type":"http","name":".*","field":".*","value":".*","headers":[]},{"type":"net","name":".*","field":".*","value":".*"},{"type":"fs","name":".*","field":".*","value":".*"},{"type":"dns","name":".*","field":".*","value":".*"}]},"payload":{"enable":"false","type":"dir","dir":"/tmp"},"tags":{},"protocol":[],"cribl":{"enable":"false","transport":{"type":"edge"},"authtoken":""}}},"environment":{}}}],
   "required": [
     "format",
     "info"
@@ -487,6 +487,9 @@
                     "enable": {
                       "$ref": "definitions/data.schema.json#/$defs/enable"
                     },
+                    "type": {
+                      "$ref": "definitions/data.schema.json#/$defs/payload_type"
+                    },
                     "dir": {
                       "$ref": "definitions/data.schema.json#/$defs/dir"
                     }

src/cfg.c

@@ -65,6 +65,7 @@ struct _config_t
 
     struct {
         unsigned int enable;
+        bool dirEnable;      // TODO: This should be modified if we decide for full transport support payload channel
         char *dir;
     } pay;
 
@@ -230,6 +231,7 @@ cfgCreateDefault(void)
     c->log.level = DEFAULT_LOG_LEVEL;
 
     c->pay.enable = DEFAULT_PAYLOAD_ENABLE;
+    c->pay.dirEnable = DEFAULT_PAYLOAD_DIR_ENABLE;
     c->pay.dir = (DEFAULT_PAYLOAD_DIR) ? scope_strdup(DEFAULT_PAYLOAD_DIR) : NULL;
 
     c->tags = DEFAULT_CUSTOM_TAGS;
@@ -622,6 +624,11 @@ cfgPayEnable(config_t *cfg)
     return (cfg) ? cfg->pay.enable : DEFAULT_PAYLOAD_ENABLE;
 }
 
+unsigned int
+cfgPayDirEnable(config_t *cfg) {
+    return (cfg) ? cfg->pay.dirEnable : DEFAULT_PAYLOAD_DIR_ENABLE;
+}
+
 const char *
 cfgPayDir(config_t *cfg)
 {
@@ -1009,6 +1016,13 @@ cfgPayEnableSet(config_t *cfg, unsigned int val)
     cfg->pay.enable = val;
 }
 
+void
+cfgPayDirEnableSet(config_t *cfg, unsigned int val)
+{
+    if (!cfg || val > 1) return;
+    cfg->pay.dirEnable = val;
+}
+
 void
 cfgPayDirSet(config_t *cfg, const char *dir)
 {

src/cfg.h

@@ -45,6 +45,7 @@ custom_tag_t**      cfgCustomTags(config_t*);
 const char*         cfgCustomTagValue(config_t*, const char*);
 cfg_log_level_t     cfgLogLevel(config_t*);
 unsigned int        cfgPayEnable(config_t*);
+unsigned int        cfgPayDirEnable(config_t *);
 const char *        cfgPayDir(config_t*);
 const char *        cfgEvtFormatHeader(config_t *, int);
 unsigned            cfgEvtAllowBinaryConsole(config_t *);
@@ -85,6 +86,7 @@ void                cfgTransportTlsCACertPathSet(config_t *, which_transport_t,
 void                cfgCustomTagAdd(config_t*, const char*, const char*);
 void                cfgLogLevelSet(config_t*, cfg_log_level_t);
 void                cfgPayEnableSet(config_t*, unsigned int);
+void                cfgPayDirEnableSet(config_t *, unsigned int);
 void                cfgPayDirSet(config_t*, const char *);
 void                cfgEvtFormatHeaderSet(config_t *, const char *);
 void                cfgEvtAllowBinaryConsoleSet(config_t *, unsigned);

src/cfgutils.c

@@ -73,6 +73,7 @@
 
 #define PAYLOAD_NODE         "payload"
 #define ENABLE_NODE              "enable"
+#define TYPE_NODE                "type"
 #define DIR_NODE                 "dir"
 
 #define CRIBL_NODE           "cribl"
@@ -167,6 +168,12 @@ enum_map_t boolMap[] = {
     {NULL,                    -1}
 };
 
+enum_map_t payTypeMap[] = {
+    {"dir",                   TRUE},
+    {"event",                 FALSE},
+    {NULL,                    -1}
+};
+
 // forward declarations
 void cfgMtcEnableSetFromStr(config_t*, const char*);
 void cfgMtcFormatSetFromStr(config_t*, const char*);
@@ -194,6 +201,7 @@ void cfgCustomTagAddFromStr(config_t*, const char*, const char*);
 void cfgLogLevelSetFromStr(config_t*, const char*);
 void cfgPayEnableSetFromStr(config_t*, const char*);
 void cfgPayDirSetFromStr(config_t*, const char*);
+void cfgPayTypeSetFromStr(config_t *, const char *);
 void cfgAuthTokenSetFromStr(config_t*, const char*);
 void cfgEvtFormatHeaderSetFromStr(config_t *, const char *);
 void cfgCriblEnableSetFromStr(config_t *, const char *);
@@ -530,6 +538,8 @@ processEnvStyleInput(config_t *cfg, const char *env_line)
         cfgTransportTlsCACertPathSetFromStr(cfg, CFG_LOG, value);
     } else if (!scope_strcmp(env_name, "SCOPE_PAYLOAD_ENABLE")) {
         cfgPayEnableSetFromStr(cfg, value);
+    } else if (!scope_strcmp(env_name, "SCOPE_PAYLOAD_DEST")) {
+        cfgPayTypeSetFromStr(cfg, value);
     } else if (!scope_strcmp(env_name, "SCOPE_PAYLOAD_DIR")) {
         cfgPayDirSetFromStr(cfg, value);
     } else if (!scope_strcmp(env_name, "SCOPE_CMD_DBG_PATH")) {
@@ -957,6 +967,18 @@ cfgPayDirSetFromStr(config_t *cfg, const char *value)
     cfgPayDirSet(cfg, value);
 }
 
+void
+cfgPayTypeSetFromStr(config_t *cfg, const char *value)
+{
+    if (!cfg || !value) return;
+    // see if value equals "dir"/"event"
+    if (scope_strncmp(value, "event", C_STRLEN("event")) == 0) {
+        cfgPayDirEnableSet(cfg, FALSE);
+    } else if (scope_strncmp(value, "dir", C_STRLEN("dir")) == 0) {
+        cfgPayDirEnableSet(cfg, TRUE);
+    }
+}
+
 void
 cfgCriblEnableSetFromStr(config_t *cfg, const char *value)
 {
@@ -1629,6 +1651,14 @@ processPayloadEnable(config_t *config, yaml_document_t *doc, yaml_node_t *node)
     if (value) scope_free(value);
 }
 
+static void
+processPayloadType(config_t *config, yaml_document_t *doc, yaml_node_t *node)
+{
+    char* value = stringVal(node);
+    cfgPayTypeSetFromStr(config, value);
+    if (value) scope_free(value);
+}
+
 static void
 processPayloadDir(config_t *config, yaml_document_t *doc, yaml_node_t *node)
 {
@@ -1644,6 +1674,7 @@ processPayload(config_t *config, yaml_document_t *doc, yaml_node_t *node)
 
     parse_table_t t[] = {
         {YAML_SCALAR_NODE,    ENABLE_NODE,          processPayloadEnable},
+        {YAML_SCALAR_NODE,    TYPE_NODE,            processPayloadType},
         {YAML_SCALAR_NODE,    DIR_NODE,             processPayloadDir},
         {YAML_NO_NODE,        NULL,                 NULL}
     };
@@ -2560,6 +2591,8 @@ createPayloadJson(config_t *cfg)
 
     if (!cJSON_AddStringToObjLN(root, ENABLE_NODE,
                          valToStr(boolMap, cfgPayEnable(cfg)))) goto err;
+    if (!cJSON_AddStringToObjLN(root, TYPE_NODE,
+                         valToStr(payTypeMap, cfgPayDirEnable(cfg)))) goto err;    
     if (!cJSON_AddStringToObjLN(root, DIR_NODE,
                          cfgPayDir(cfg))) goto err;
 
@@ -2863,10 +2896,14 @@ initCtl(config_t *cfg)
      */
     payload_status_t payloadStatus = PAYLOAD_STATUS_DISABLE;
     if (cfgPayEnable(cfg) || protocolDefinitionsUsePayloads()) {
-        if (cfgLogStreamEnable(cfg) && !payloadToDiskForced() ) {
-            payloadStatus = PAYLOAD_STATUS_CRIBL;
-        } else {
-            payloadStatus = PAYLOAD_STATUS_DISK;
+        payloadStatus = PAYLOAD_STATUS_DISK;
+        bool payloadOnDisk = cfgPayDirEnable(cfg);
+        if (payloadOnDisk == FALSE) {
+            if (cfgLogStreamEnable(cfg)) {
+                payloadStatus = PAYLOAD_STATUS_CRIBL;
+            } else if (cfgEvtEnable(cfg)) {
+                payloadStatus = PAYLOAD_STATUS_CTL;
+            }
         }
     }
     if (payloadStatus == PAYLOAD_STATUS_CRIBL) {