Merge pull request #122 from ddosify/feat/filter-processes

Feat/filter processes

.gitignore

@@ -1 +1,3 @@
-.DS_Store
+.DS_Store
+.vscode/settings.json
+.vscode/launch.json

Dockerfile.default

@@ -1,10 +1,12 @@
-FROM golang:1.20.14-bullseye as builder
+FROM golang:1.22.1-bullseye as builder
 WORKDIR /app
 COPY . ./
 RUN apt update 
 
 ARG VERSION
-RUN GOOS=linux go build -ldflags="-X 'github.com/ddosify/alaz/datastore.tag=$VERSION'" -o alaz
+ENV GOCACHE=/root/.cache/go-build
+RUN go mod tidy -v
+RUN --mount=type=cache,target="/root/.cache/go-build" GOOS=linux go build -ldflags="-X 'github.com/ddosify/alaz/datastore.tag=$VERSION'" -o alaz
 
 FROM debian:12.5-slim
 RUN apt-get update && apt-get install -y procps ca-certificates && rm -rf /var/lib/apt/lists/*

Makefile

@@ -18,7 +18,7 @@ CONTAINER_ENGINE ?= docker
 CONTAINER_RUN_ARGS ?= $(--user "${UIDGID}")
 
 IMAGE_GENERATE := ebpf-builder
-VERSION_GENERATE := v1
+VERSION_GENERATE := v1-1.22.1
 GENERATE_DOCKERFILE := ebpf-builder/Dockerfile
 
 # clang <8 doesn't tag relocs properly (STT_NOTYPE)

aggregator/data.go

@@ -1016,7 +1016,7 @@ func (a *Aggregator) processL7(ctx context.Context, d *l7_req.L7Event) {
 	if skInfo == nil {
 		log.Logger.Debug().Uint32("pid", d.Pid).
 			Uint64("fd", d.Fd).Uint64("writeTime", d.WriteTimeNs).
-			Str("protocol", d.Protocol).Any("payload", string(d.Payload[:])).Msg("socket not found")
+			Str("protocol", d.Protocol).Any("payload", string(d.Payload[:d.PayloadSize])).Msg("socket not found")
 		return
 	}
 

cri/cri.go

@@ -1,12 +1,18 @@
 package cri
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
+	"regexp"
+	"strconv"
 	"time"
 
 	"github.com/ddosify/alaz/log"
+	"github.com/prometheus/procfs"
+
 	//nolint:staticcheck
 	//nolint:staticcheck
 	internalapi "k8s.io/cri-api/pkg/apis"
@@ -26,7 +32,8 @@ type ContainerPodInfo struct {
 }
 
 type CRITool struct {
-	rs internalapi.RuntimeService
+	rs         internalapi.RuntimeService
+	nsFilterRx *regexp.Regexp
 }
 
 func NewCRITool(ctx context.Context) (*CRITool, error) {
@@ -47,11 +54,42 @@ func NewCRITool(ctx context.Context) (*CRITool, error) {
 		return nil, err
 	}
 
+	var nsFilterRx *regexp.Regexp
+	if os.Getenv("EXCLUDE_NAMESPACES") != "" {
+		nsFilterRx = regexp.MustCompile(os.Getenv("EXCLUDE_NAMESPACES"))
+	}
+
 	return &CRITool{
-		rs: res,
+		rs:         res,
+		nsFilterRx: nsFilterRx,
 	}, nil
 }
 
+func (ct *CRITool) FilterNamespace(ns string) bool {
+	if ns == "kube-system" {
+		return true
+	}
+	if ct.nsFilterRx == nil {
+		return false
+	}
+	if ct.nsFilterRx.MatchString(ns) {
+		log.Logger.Debug().Msgf("%s filtered with EXCLUDE_NAMESPACES regex %s", ns, ct.nsFilterRx.String())
+		return true
+	}
+
+	return false
+}
+
+func (ct *CRITool) FilterNamespaceWithContainerId(id string) bool {
+	resp, err := ct.ContainerStatus(id)
+	if err != nil {
+		log.Logger.Error().Err(err).Msgf("Failed to get container status for container %s", id)
+		return true
+	}
+
+	return ct.FilterNamespace(resp.PodNs)
+}
+
 func (ct *CRITool) GetAllContainers() ([]*pb.Container, error) {
 	// get running containers
 	st := &pb.ContainerStateValue{}
@@ -74,6 +112,119 @@ func (ct *CRITool) GetAllContainers() ([]*pb.Container, error) {
 	return list, nil
 }
 
+func (ct *CRITool) GetPidsRunningOnContainers() (map[uint32]struct{}, error) {
+	pids := make(map[uint32]struct{})
+	st := &pb.ContainerStateValue{}
+	st.State = pb.ContainerState_CONTAINER_RUNNING
+
+	filter := &pb.ContainerFilter{
+		Id:                   "",
+		State:                st,
+		PodSandboxId:         "",
+		LabelSelector:        map[string]string{},
+		XXX_NoUnkeyedLiteral: struct{}{},
+		XXX_sizecache:        0,
+	}
+
+	list, err := ct.rs.ListContainers(context.TODO(), filter)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, c := range list {
+		if ct.FilterNamespaceWithContainerId(c.Id) {
+			log.Logger.Debug().Msgf("No tracking on ebpf side for container [%s] - [%s]", c.Id, c.Metadata.Name)
+			continue
+		}
+		runningPids, err := ct.getAllRunningProcsInsideContainer(c.Id)
+		if err != nil {
+			log.Logger.Error().Err(err).Msgf("Failed to get runnning pids for container [%s]", c.Id)
+			continue
+		}
+		log.Logger.Debug().Msgf("running container [%s-%s] has pids %v", c.Metadata.Name, c.Id, runningPids)
+
+		for _, pid := range runningPids {
+			pids[pid] = struct{}{}
+		}
+	}
+	return pids, nil
+}
+
+func (ct *CRITool) getAllRunningProcsInsideContainer(containerID string) ([]uint32, error) {
+	r, err := ct.rs.ContainerStatus(context.TODO(), containerID, true)
+	if err != nil {
+		log.Logger.Error().Err(err).Msgf("Failed to get container status for container %s", containerID)
+		return nil, err
+	}
+
+	info := map[string]interface{}{}
+	json.Unmarshal([]byte(r.Info["info"]), &info)
+
+	// pid of main process
+	pidFloat := info["pid"].(float64)
+	pid := int(pidFloat)
+
+	fs, err := procfs.NewFS("/proc/1/root/proc")
+	if err != nil {
+		return nil, err
+	}
+
+	proc, err := fs.Proc(pid)
+	if err != nil {
+		return nil, err
+	}
+
+	// get cgroup hiearchies and read cgrou.procs file for pids
+	cgroups, err := proc.Cgroups()
+	if err != nil {
+		return nil, err
+	}
+
+	result := []uint32{}
+
+	for _, cgroup := range cgroups {
+		if cgroup.HierarchyID == 0 { // cgroup v2
+			procsPath := "/proc/1/root/sys/fs/cgroup" + cgroup.Path + "/cgroup.procs"
+			// read proc pids
+			pidFile, err := os.OpenFile(procsPath, os.O_RDONLY, 0)
+			if err != nil {
+				log.Logger.Warn().Err(err).Msgf("Error reading cgroup.procs file for cgroup %s", cgroup.Path)
+				continue
+			}
+			defer pidFile.Close()
+			fileScanner := bufio.NewScanner(pidFile)
+			for fileScanner.Scan() {
+				pid, err := strconv.ParseUint(fileScanner.Text(), 10, 32)
+				if err != nil {
+					log.Logger.Warn().Err(err).Msgf("Error parsing pid %s", fileScanner.Text())
+					continue
+				}
+				result = append(result, uint32(pid))
+			}
+		} else { // v1 cgroup
+			// use memory controller as default
+			procsPath := "/proc/1/root/sys/fs/cgroup/memory" + cgroup.Path + "/cgroup.procs"
+			// read proc pids
+			pidFile, err := os.OpenFile(procsPath, os.O_RDONLY, 0)
+			if err != nil {
+				log.Logger.Warn().Err(err).Msgf("Error reading cgroup.procs file for cgroup %s", cgroup.Path)
+				continue
+			}
+			defer pidFile.Close()
+			fileScanner := bufio.NewScanner(pidFile)
+			for fileScanner.Scan() {
+				pid, err := strconv.ParseUint(fileScanner.Text(), 10, 32)
+				if err != nil {
+					log.Logger.Warn().Err(err).Msgf("Error parsing pid %s", fileScanner.Text())
+					continue
+				}
+				result = append(result, uint32(pid))
+			}
+		}
+	}
+	return result, nil
+}
+
 // get log path of container
 // id string : containerID
 func (ct *CRITool) GetLogPath(id string) (string, error) {

datastore/backend.go

@@ -834,7 +834,7 @@ func (b *BackendDS) PersistContainer(c Container, eventType string) error {
 	return nil
 }
 
-func (b *BackendDS) SendHealthCheck(ebpf bool, metrics bool, dist bool, k8sVersion string) {
+func (b *BackendDS) SendHealthCheck(tracing bool, metrics bool, logs bool, k8sVersion string) {
 	t := time.NewTicker(10 * time.Second)
 	defer t.Stop()
 
@@ -847,13 +847,13 @@ func (b *BackendDS) SendHealthCheck(ebpf bool, metrics bool, dist bool, k8sVersi
 				AlazVersion:    tag,
 			},
 			Info: struct {
-				EbpfEnabled        bool `json:"ebpf"`
-				MetricsEnabled     bool `json:"metrics"`
-				DistTracingEnabled bool `json:"traffic"`
+				TracingEnabled bool `json:"tracing"`
+				MetricsEnabled bool `json:"metrics"`
+				LogsEnabled    bool `json:"logs"`
 			}{
-				EbpfEnabled:        ebpf,
-				MetricsEnabled:     metrics,
-				DistTracingEnabled: dist,
+				TracingEnabled: tracing,
+				MetricsEnabled: metrics,
+				LogsEnabled:    logs,
 			},
 			Telemetry: struct {
 				KernelVersion string `json:"kernel_version"`

datastore/payload.go

@@ -10,9 +10,9 @@ type Metadata struct {
 type HealthCheckPayload struct {
 	Metadata Metadata `json:"metadata"`
 	Info     struct {
-		EbpfEnabled        bool `json:"ebpf"`
-		MetricsEnabled     bool `json:"metrics"`
-		DistTracingEnabled bool `json:"traffic"`
+		TracingEnabled bool `json:"tracing"`
+		MetricsEnabled bool `json:"metrics"`
+		LogsEnabled    bool `json:"logs"`
 	} `json:"alaz_info"`
 	Telemetry struct {
 		KernelVersion string `json:"kernel_version"`

ebpf-builder/Dockerfile

@@ -1,7 +1,7 @@
 FROM ubuntu:22.04
 
 ENV LIBBPF_VERSION 1.2.2
-ENV GOLANG_VERSION 1.20.5
+ENV GOLANG_VERSION 1.22.1
 
 # Install Clang and LLVM Strip
 RUN apt-get update && apt-get install -y clang-14 llvm && \

ebpf/bpf.go

@@ -3,7 +3,6 @@ package ebpf
 import "context"
 
 type Program interface {
-	Load()                                            // Load bpf program to kernel
 	Attach()                                          // attach links to programs, in case error process must exit
 	InitMaps()                                        // initialize bpf map readers, must be called before Consume
 	Consume(ctx context.Context, ch chan interface{}) // consume bpf events, publishes to chan provided

ebpf/c/bpf.c

@@ -0,0 +1,44 @@
+//go:build ignore
+
+#include "../headers/bpf.h"
+#include "../headers/common.h"
+#include "../headers/tcp.h"
+#include "../headers/l7_req.h"
+
+
+// order is important
+#ifndef __BPF__H
+#define __BPF__H
+#include <bpf/bpf_core_read.h>
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_endian.h>
+#include <bpf/bpf_tracing.h>
+#endif
+
+#define FILTER_OUT_NON_CONTAINER
+
+#include <stddef.h>
+#include "../headers/pt_regs.h"
+#include <sys/socket.h>
+
+#include "../headers/log.h"
+
+#include "macros.h"
+#include "struct.h"
+#include "map.h"
+
+#include "tcp.c"
+#include "proc.c"
+
+#include "http.c"
+#include "amqp.c"
+#include "postgres.c"
+#include "openssl.c"
+#include "http2.c"
+#include "tcp_sock.c"
+#include "go_internal.h"
+#include "l7.c"
+
+char __license[] SEC("license") = "Dual MIT/GPL";
+
+