Add support for payload destination in CLI/cfg

- allows to change payload destination to event transport
- by default the payloads will go to directory
- it is possible to change this behaviour via:
  - `--payloadsdest` argument in `scope` CLI
  - `SCOPE_PAYLOAD_DEST` environment variable
  - allowed values are for the above are "event", "dir"

Closes #1594

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 rc.PayloadsDest != "" && 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

@@ -86,6 +86,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 rc.PayloadsDest != "" && 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 rc.PayloadsDest != "" && 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

@@ -191,10 +191,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 = "event"
+		// 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)

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 t 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,8 +2896,13 @@ initCtl(config_t *cfg)
      */
     payload_status_t payloadStatus = PAYLOAD_STATUS_DISABLE;
     if (cfgPayEnable(cfg) || protocolDefinitionsUsePayloads()) {
-        if (cfgLogStreamEnable(cfg) && !payloadToDiskForced() ) {
-            payloadStatus = PAYLOAD_STATUS_CRIBL;
+        bool payloadOnDisk = cfgPayDirEnable(cfg);
+        if (payloadOnDisk == FALSE) {
+            if (cfgLogStreamEnable(cfg)) {
+                payloadStatus = PAYLOAD_STATUS_CRIBL;
+            } else if (cfgEvtEnable(cfg)) {
+                payloadStatus = PAYLOAD_STATUS_CTL;
+            }
         } else {
             payloadStatus = PAYLOAD_STATUS_DISK;
         }

src/ctl.c

@@ -1232,33 +1232,35 @@ ctlEvtGet(ctl_t *ctl)
     return ctl ? ctl->evt : NULL;
 }
 
-static transport_status_t
-ctlPayConnectionStatus(ctl_t * ctl, payload_status_t payStatus) {
+transport_status_t
+ctlPayloadConnectionStatus(ctl_t *ctl) {
+    // retrieve the information about source 
+    payload_status_t payStatus = ctlPayStatus(ctl);
     transport_status_t status = {
         .configString = NULL,
         .isConnected = FALSE,
         .connectAttemptCount = 0,
         .failureString = NULL};
 
-    if ((!ctl) || (payStatus != PAYLOAD_STATUS_DISK)) {
-        return status;
+    switch (payStatus) {
+        case PAYLOAD_STATUS_DISABLE:
+            return status;
+        case PAYLOAD_STATUS_CRIBL:
+            return transportConnectionStatus(ctl->paytrans);
+        case PAYLOAD_STATUS_CTL:
+            return transportConnectionStatus(ctl->transport);
+        case PAYLOAD_STATUS_DISK:
+            status.configString = ctl->payload.dirRepr;
+            status.isConnected = TRUE;
+            return status;
+        default:
+            DBG(NULL);
+            return status;
     }
-
-    status.configString = ctl->payload.dirRepr;
-    status.isConnected = TRUE;
-
-    return status;
 }
 
 transport_status_t
-ctlConnectionStatus(ctl_t *ctl, which_transport_t who) {
-    if (who == CFG_LS) {
-        payload_status_t status = ctlPayStatus(ctl);
-        if (status == PAYLOAD_STATUS_CRIBL) {
-            return transportConnectionStatus(ctl->paytrans);
-        }
-        return ctlPayConnectionStatus(ctl, status);
-    }
+ctlConnectionStatus(ctl_t *ctl) {
     return transportConnectionStatus(ctl->transport);
 }
 

src/ctl.h

@@ -36,8 +36,9 @@ typedef enum {
 
 typedef enum {
     PAYLOAD_STATUS_DISABLE = 0,    // payloads are disabled
-    PAYLOAD_STATUS_CRIBL = 1,      // payloads are enabled and will go to cribl
-    PAYLOAD_STATUS_DISK = 2,       // payloads are enabled and will go on disk
+    PAYLOAD_STATUS_CRIBL = 1,      // payloads are enabled and will use cribl transport
+    PAYLOAD_STATUS_CTL = 2,        // payloads are enabled and will use event transport
+    PAYLOAD_STATUS_DISK = 3,       // payloads are enabled and will go on disk
 } payload_status_t;
 
 /**
@@ -133,7 +134,8 @@ void                ctlTransportSet(ctl_t *, transport_t *, which_transport_t);
 transport_t *       ctlTransport(ctl_t *, which_transport_t);
 evt_fmt_t *         ctlEvtGet(ctl_t *);
 void                ctlEvtSet(ctl_t *, evt_fmt_t *);
-transport_status_t  ctlConnectionStatus(ctl_t *, which_transport_t);
+transport_status_t  ctlConnectionStatus(ctl_t *);
+transport_status_t  ctlPayloadConnectionStatus(ctl_t *);
 
 // Accessor for performance
 bool            ctlEvtSourceEnabled(ctl_t *, watch_t);

src/ipc_resp.c

@@ -327,7 +327,7 @@ payloadTransportEnabled(void) {
  */
 static transport_status_t
 payloadTransportStatus(void) {
-    return ctlConnectionStatus(g_ctl, CFG_LS);
+    return ctlPayloadConnectionStatus(g_ctl);
 }
 
 /*
@@ -351,7 +351,7 @@ eventsTransportEnabled(void) {
  */
 static transport_status_t
 eventsTransportStatus(void) {
-    return ctlConnectionStatus(g_ctl, CFG_CTL);
+    return ctlConnectionStatus(g_ctl);
 }
 
 /*