Add initial process tree support

The process tree gives you context about running processes at a given
time. They are used by the cast backend to provide more context to the
user.

Author: patrickpichler

Tiltfile

@@ -26,6 +26,9 @@ local_resource(
         './pkg/cgroup',
         './pkg/containers',
         './pkg/bucketcache',
+        './pkg/processtree',
+        './pkg/proc',
+        './pkg/system',
     ],
 )
 

api/v1/runtime/runtime_agent_api.pb.go

@@ -14,6 +14,7 @@ service RuntimeSecurityAgentAPI {
   rpc LogsWriteStream(stream LogEvent) returns (WriteStreamResponse) {}
   rpc ContainerStatsWriteStream(stream ContainerStatsBatch) returns (WriteStreamResponse) {}
   rpc NetflowWriteStream(stream Netflow) returns (WriteStreamResponse) {}
+  rpc ProcessEventsWriteStream(stream ProcessTreeEvent) returns (WriteStreamResponse) {}
 
   rpc GetSyncState(GetSyncStateRequest) returns (GetSyncStateResponse) {}
   rpc UpdateSyncState(UpdateSyncStateRequest) returns (UpdateSyncStateResponse) {}
@@ -359,3 +360,32 @@ message KubeLinterCheck {
   uint64 passed = 2; // Represented as bitmasks of passed checks.
   uint64 failed = 3; // Represented as bitmasks of failed checks.
 }
+
+message Process {
+  uint32 pid = 1;
+  uint32 ppid = 2;
+  uint64 start_time = 3;
+  uint64 parent_start_time = 4;
+  repeated string args = 5;
+  string filepath = 6;
+  uint64 exit_time = 7;
+}
+
+enum ProcessAction {
+  PROCESS_ACTION_UNKNOWN = 0;
+  PROCESS_ACTION_EXEC = 1;
+  PROCESS_ACTION_FORK = 2;
+  PROCESS_ACTION_EXIT = 3;
+}
+
+message ProcessEvent {
+  uint64 timestamp = 1; // Stored as unix timestamp in nanoseconds.
+  string container_id = 2;
+  Process process = 3;
+  ProcessAction action = 4;
+}
+
+message ProcessTreeEvent {
+  bool initial = 1;
+  repeated ProcessEvent events = 2;
+}

api/v1/runtime/runtime_agent_api.proto

@@ -25,6 +25,7 @@ agent:
     netflow-sample-submit-interval-seconds: 1
     netflow-export-interval: 5s
     ebpf-events-stdio-exporter-enabled: false
+    process-tree-enabled: true
 
   prometheusScrape:
     enabled: true
@@ -51,6 +52,8 @@ controller:
 
   containerSecurityContext:
     readOnlyRootFilesystem: false
+  securityContext:
+    runAsNonRoot: false
 
   prometheusScrape:
     enabled: true

api/v1/runtime/runtime_agent_api_grpc.pb.go

@@ -28,6 +28,7 @@ import (
 	"github.com/castai/kvisor/pkg/ebpftracer/types"
 	"github.com/castai/kvisor/pkg/logging"
 	"github.com/castai/kvisor/pkg/proc"
+	"github.com/castai/kvisor/pkg/processtree"
 	"github.com/go-playground/validator/v10"
 	"github.com/grafana/pyroscope-go"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -59,6 +60,7 @@ type Config struct {
 	Castai                         castai.Config
 	EnricherConfig                 EnricherConfig
 	Netflow                        NetflowConfig
+	ProcessTree                    ProcessTreeConfig
 	Clickhouse                     ClickhouseConfig
 	KubeAPIServiceAddr             string
 	ExportersQueueSize             int `validate:"required"`
@@ -133,6 +135,10 @@ type ClickhouseConfig struct {
 	Password string
 }
 
+type ProcessTreeConfig struct {
+	Enabled bool
+}
+
 func New(cfg *Config) *App {
 	if err := validator.New().Struct(cfg); err != nil {
 		panic(fmt.Errorf("invalid config: %w", err).Error())
@@ -187,6 +193,10 @@ func (a *App) Run(ctx context.Context) error {
 		if cfg.Netflow.Enabled {
 			exporters.Netflow = append(exporters.Netflow, state.NewCastaiNetflowExporter(log, castaiClient, a.cfg.ExportersQueueSize))
 		}
+		if cfg.ProcessTree.Enabled {
+			exporter := state.NewCastaiProcessTreeExporter(log, castaiClient, a.cfg.ExportersQueueSize)
+			exporters.ProcessTree = append(exporters.ProcessTree, exporter)
+		}
 	} else {
 		log = logging.New(logCfg)
 		exporters = state.NewExporters(log)
@@ -224,6 +234,11 @@ func (a *App) Run(ctx context.Context) error {
 			clickhouseNetflowExporter := state.NewClickhouseNetflowExporter(log, storageConn, a.cfg.ExportersQueueSize)
 			exporters.Netflow = append(exporters.Netflow, clickhouseNetflowExporter)
 		}
+
+		if cfg.ProcessTree.Enabled {
+			exporter := state.NewClickhouseProcessTreeExporter(log, storageConn, a.cfg.ExportersQueueSize)
+			exporters.ProcessTree = append(exporters.ProcessTree, exporter)
+		}
 	}
 
 	if cfg.EBPFEventsStdioExporterEnabled {
@@ -245,10 +260,19 @@ func (a *App) Run(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	containersClient, err := containers.NewClient(log, cgroupClient, a.cfg.ContainerdSockPath)
+	procHandler := proc.New()
+	containersClient, err := containers.NewClient(log, cgroupClient, a.cfg.ContainerdSockPath, procHandler)
 	if err != nil {
 		return err
 	}
+	processTreeCollector, err := processtree.New(log, procHandler, containersClient)
+	if err != nil {
+		return fmt.Errorf("process tree: %w", err)
+	}
+	err = processTreeCollector.Init(ctx)
+	if err != nil {
+		return fmt.Errorf("process tree: %w", err)
+	}
 	ct, err := conntrack.NewClient(log)
 	if err != nil {
 		return fmt.Errorf("conntrack: %w", err)
@@ -262,7 +286,6 @@ func (a *App) Run(ctx context.Context) error {
 
 	signatureEngine := signature.NewEngine(activeSignatures, log, a.cfg.SignatureEngineConfig)
 
-	procHandler := proc.New()
 	mountNamespacePIDStore, err := getInitializedMountNamespaceStore(procHandler)
 	if err != nil {
 		return fmt.Errorf("mount namespace PID store: %w", err)
@@ -292,6 +315,7 @@ func (a *App) Run(ctx context.Context) error {
 		NetflowSampleSubmitIntervalSeconds: a.cfg.Netflow.SampleSubmitIntervalSeconds,
 		NetflowGrouping:                    a.cfg.Netflow.Grouping,
 		TrackSyscallStats:                  cfg.ContainerStatsEnabled,
+		ProcessTreeCollector:               processTreeCollector,
 	})
 	if err := tracer.Load(); err != nil {
 		return fmt.Errorf("loading tracer: %w", err)
@@ -300,8 +324,11 @@ func (a *App) Run(ctx context.Context) error {
 
 	policy := &ebpftracer.Policy{
 		SystemEvents: []events.ID{
-			events.SignalCgroupMkdir,
-			events.SignalCgroupRmdir,
+			events.CgroupMkdir,
+			events.CgroupRmdir,
+			events.SchedProcessExec,
+			events.SchedProcessExit,
+			events.SchedProcessFork,
 		},
 		Events: []*ebpftracer.EventPolicy{},
 	}
@@ -366,6 +393,7 @@ func (a *App) Run(ctx context.Context) error {
 		signatureEngine,
 		enrichmentService,
 		kubeAPIServerClient,
+		processTreeCollector,
 	)
 
 	errg, ctx := errgroup.WithContext(ctx)

charts/kvisor/values-local.yaml

@@ -52,6 +52,9 @@ func NewClickhouseInitCommand() *cobra.Command {
 				if err := conn.Exec(ctx, state.ClickhouseNetflowSchema()); err != nil {
 					return fmt.Errorf("creating clickhouse netflow schema: %w", err)
 				}
+				if err := conn.Exec(ctx, state.ClickhouseProcessTreeSchema()); err != nil {
+					return fmt.Errorf("creating clickhouse process tree schema: %w", err)
+				}
 
 				return nil
 			}

cmd/agent/daemon/app/app.go

@@ -73,6 +73,8 @@ func NewRunCommand(version string) *cobra.Command {
 		netflowExportInterval              = pflag.Duration("netflow-export-interval", 15*time.Second, "Netflow export interval")
 		netflowGrouping                    = ebpftracer.NetflowGroupingDropSrcPort
 
+		processTreeEnabled                     = pflag.Bool("process-tree-enabled", false, "Enables process tree tracking")
+
 		clickhouseAddr     = pflag.String("clickhouse-addr", "", "Clickhouse address to send events to")
 		clickhouseDatabase = pflag.String("clickhouse-database", "", "Clickhouse database name")
 		clickhouseUsername = pflag.String("clickhouse-username", "", "Clickhouse username")
@@ -160,6 +162,9 @@ func NewRunCommand(version string) *cobra.Command {
 					Username: *clickhouseUsername,
 					Password: os.Getenv("CLICKHOUSE_PASSWORD"),
 				},
+        ProcessTree: app.ProcessTreeConfig{
+        	Enabled: *processTreeEnabled,
+        },
 				KubeAPIServiceAddr: *kubeAPIServiceAddr,
 				ExportersQueueSize: *exportersQueueSize,
 			}).Run(ctx); err != nil && !errors.Is(err, context.Canceled) {

cmd/agent/daemon/clickouse_init.go

@@ -43,7 +43,7 @@ func (c *CastaiEventsExporter) Run(ctx context.Context) error {
 	defer ws.Close()
 	ws.ReopenDelay = c.writeStreamCreateRetryDelay
 
-	sender := &sender{
+	sender := &eventSender{
 		ws:              ws,
 		retryQueue:      c.retryQueue,
 		sendMetric:      metrics.AgentExporterSendTotal.WithLabelValues("castai_events"),
@@ -62,15 +62,15 @@ func (c *CastaiEventsExporter) Run(ctx context.Context) error {
 	}
 }
 
-type sender struct {
+type eventSender struct {
 	ws         *castai.WriteStream[*castpb.Event, *castpb.WriteStreamResponse]
 	retryQueue chan *castpb.Event
 
 	sendMetric      prometheus.Counter
 	sendErrorMetric prometheus.Counter
 }
 
-func (s *sender) send(e *castpb.Event, retry bool) {
+func (s *eventSender) send(e *castpb.Event, retry bool) {
 	if err := s.ws.Send(e); err != nil {
 		if retry {
 			s.retryQueue <- e