diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index dc366d630..718b6b0de 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -59,7 +59,7 @@ Activate a new docker profile. You can run tests on your local kind cluster. ```sh -KIND_CONTEXT=tilt IMAGE_TAG=local ./e2e/run.sh +KIND_CONTEXT=kind IMAGE_TAG=local ./e2e/run.sh ``` You will need IPv6 to be enabled on the host. Most operating systems / distros have IPv6 enabled by default, but you can check on Linux with the following command: diff --git a/charts/kvisor/templates/agent.yaml b/charts/kvisor/templates/agent.yaml index cac6bc6b5..916dd3da6 100644 --- a/charts/kvisor/templates/agent.yaml +++ b/charts/kvisor/templates/agent.yaml @@ -110,6 +110,12 @@ spec: {{- else -}} {{ .Values.castai.grpcAddr | quote }} {{- end }} + - name: CASTAI_API_URL + value: {{ if .Values.mockServer.enabled -}} + {{ (printf "http://%s:8080" (include "kvisor.castaiMockServer.service" .)) | quote }} + {{- else -}} + {{ .Values.castai.apiURL | quote }} + {{- end }} {{- include "kvisor.clusterIDEnv" (set (deepCopy .) "envFrom" .Values.agent.envFrom) | nindent 12 }} {{- if .Values.agent.debug.ebpf }} - name: KVISOR_EBPF_DEBUG diff --git a/charts/kvisor/templates/cast-mock-server.yaml b/charts/kvisor/templates/cast-mock-server.yaml index b03b57818..755562840 100644 --- a/charts/kvisor/templates/cast-mock-server.yaml +++ b/charts/kvisor/templates/cast-mock-server.yaml @@ -54,8 +54,12 @@ spec: {{- include "kvisor.castaiMockServer.selectorLabels" . | nindent 4 }} type: ClusterIP ports: - - name: server + - name: grpc protocol: TCP port: 8443 targetPort: 8443 + - name: http + protocol: TCP + port: 8080 + targetPort: 8080 {{- end }} diff --git a/charts/kvisor/templates/controller.yaml b/charts/kvisor/templates/controller.yaml index fbeca56c3..319f0f98f 100644 --- a/charts/kvisor/templates/controller.yaml +++ b/charts/kvisor/templates/controller.yaml @@ -92,6 +92,12 @@ spec: {{- else -}} {{ .Values.castai.grpcAddr | quote }} {{- end }} + - name: CASTAI_API_URL + value: {{ if .Values.mockServer.enabled -}} + {{ (printf "http://%s:8080" (include "kvisor.castaiMockServer.service" .)) | quote }} + {{- else -}} + {{ .Values.castai.apiURL | quote }} + {{- end }} {{- include "kvisor.clusterIDEnv" (set (deepCopy .) "envFrom" .Values.controller.envFrom) | nindent 12 }} {{- range $k, $v := .Values.controller.additionalEnv }} - name: {{ $k }} diff --git a/charts/kvisor/values.yaml b/charts/kvisor/values.yaml index 865347192..27820897c 100644 --- a/charts/kvisor/values.yaml +++ b/charts/kvisor/values.yaml @@ -17,6 +17,8 @@ castai: # Note: If your cluster is in the EU region, update the grpcAddr to: https://kvisor.prod-eu.cast.ai:443 grpcAddr: "kvisor.prod-master.cast.ai:443" + apiURL: "" + # clusterID and clusterIdSecretKeyRef are mutually exclusive clusterID: "" # clusterIdSecretKeyRef -- Name and Key of secret with ClusterID diff --git a/cmd/agent/daemon/app/app.go b/cmd/agent/daemon/app/app.go index 6a8158d96..dd0a998e0 100644 --- a/cmd/agent/daemon/app/app.go +++ b/cmd/agent/daemon/app/app.go @@ -21,7 +21,6 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/samber/lo" "golang.org/x/sync/errgroup" - "golang.org/x/time/rate" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -44,10 +43,10 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/signature" "github.com/castai/kvisor/pkg/ebpftracer/types" "github.com/castai/kvisor/pkg/kernel" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" "github.com/castai/kvisor/pkg/processtree" - castlog "github.com/castai/logging" + "github.com/castai/logging" + "github.com/castai/logging/components" custommetrics "github.com/castai/metrics" ) @@ -65,20 +64,15 @@ type App struct { func (a *App) Run(ctx context.Context) error { start := time.Now() - cfg := a.cfg - logCfg := &logging.Config{ - Level: logging.MustParseLevel(a.cfg.LogLevel), - AddSource: true, - RateLimiter: logging.RateLimiterConfig{ - Limit: rate.Every(a.cfg.LogRateInterval), - Burst: a.cfg.LogRateBurst, - Inform: true, - }, - } + errg, ctx := errgroup.WithContext(ctx) + cfg := a.cfg podName := os.Getenv("POD_NAME") var log *logging.Logger + logHandlers := []logging.Handler{ + logging.NewTextHandler(logging.DefaultTextHandlerConfig), + } var exporters []export.DataBatchWriter // Castai specific spetup if config is valid. if cfg.Castai.Valid() { @@ -90,28 +84,40 @@ func (a *App) Run(ctx context.Context) error { return fmt.Errorf("sync remote config: %w", err) } if cfg.SendLogsLevel != "" { - castaiLogsExporter := castai.NewLogsExporter(castaiClient) - go castaiLogsExporter.Run(ctx) //nolint:errcheck + logsApiClient, err := components.NewAPIClient(components.Config{ + APIBaseURL: a.cfg.Castai.APIURL, + APIKey: a.cfg.Castai.APIKey, + ClusterID: a.cfg.Castai.ClusterID, + Component: "kvisor-agent", + Version: a.cfg.Version, + }) + if err != nil { + return fmt.Errorf("creating logs api client: %w", err) + } + batchLogsApiClient := components.NewBatchClient(logsApiClient) + errg.Go(func() error { + return batchLogsApiClient.Run(ctx) + }) + logsExportHandler := logging.NewExportHandler(batchLogsApiClient, logging.ExportHandlerConfig{ + MinLevel: logging.MustParseLevel(cfg.SendLogsLevel), + }) + logHandlers = append(logHandlers, logsExportHandler) if cfg.PromMetricsExportEnabled { - castaiMetricsExporter := castai.NewPromMetricsExporter(log, castaiLogsExporter, prometheus.DefaultGatherer, castai.PromMetricsExporterConfig{ + castaiMetricsExporter := castai.NewPromMetricsExporter(log, batchLogsApiClient, prometheus.DefaultGatherer, castai.PromMetricsExporterConfig{ PodName: podName, ExportInterval: cfg.PromMetricsExportInterval, }) - go castaiMetricsExporter.Run(ctx) //nolint:errcheck - } - - logCfg.Export = logging.ExportConfig{ - ExportFunc: castaiLogsExporter.ExportFunc(), - MinLevel: logging.MustParseLevel(cfg.SendLogsLevel), + errg.Go(func() error { + return castaiMetricsExporter.Run(ctx) + }) } } - log = logging.New(logCfg) - + log = logging.New(logHandlers...) exporters = append(exporters, castaiexport.NewDataBatchWriter(castaiClient, log)) } else { - log = logging.New(logCfg) + log = logging.New(logHandlers...) log.Warn("castai config is not set or it is invalid, running agent in standalone mode") } @@ -263,7 +269,7 @@ func (a *App) Run(ctx context.Context) error { var cloudVolumeMetricsWriter pipeline.CloudVolumeMetricsWriter var storageInfoProvider pipeline.StorageInfoProvider if cfg.Stats.StorageEnabled { - metricsClient, err := createMetricsClient(cfg) + metricsClient, err := createMetricsClient(cfg, log) if err != nil { return fmt.Errorf("failed to create metrics client: %w", err) } @@ -323,7 +329,6 @@ func (a *App) Run(ctx context.Context) error { } } - errg, ctx := errgroup.WithContext(ctx) errg.Go(func() error { return a.runHTTPServer(ctx, log) }) @@ -648,7 +653,7 @@ func resolveMetricsAddr(addr string) string { return addr } -func createMetricsClient(cfg *config.Config) (custommetrics.MetricClient, error) { +func createMetricsClient(cfg *config.Config, log *logging.Logger) (custommetrics.MetricClient, error) { if !cfg.Castai.Valid() { return nil, fmt.Errorf("cast config is not valid") } @@ -660,7 +665,7 @@ func createMetricsClient(cfg *config.Config) (custommetrics.MetricClient, error) Insecure: cfg.Castai.Insecure, } - return custommetrics.NewMetricClient(metricsClientConfig, castlog.New()) + return custommetrics.NewMetricClient(metricsClientConfig, log) } type noopTracer struct{} diff --git a/cmd/agent/daemon/clickhouse_init.go b/cmd/agent/daemon/clickhouse_init.go index 130f27a9c..4f963bfa4 100644 --- a/cmd/agent/daemon/clickhouse_init.go +++ b/cmd/agent/daemon/clickhouse_init.go @@ -9,9 +9,10 @@ import ( "time" "github.com/ClickHouse/clickhouse-go/v2" - clickhouseexport "github.com/castai/kvisor/cmd/agent/daemon/export/clickhouse" - "github.com/castai/kvisor/pkg/logging" "github.com/spf13/cobra" + + clickhouseexport "github.com/castai/kvisor/cmd/agent/daemon/export/clickhouse" + "github.com/castai/logging" ) func NewClickhouseInitCommand() *cobra.Command { @@ -24,7 +25,7 @@ func NewClickhouseInitCommand() *cobra.Command { command := &cobra.Command{ Use: "clickhouse-init", Run: func(cmd *cobra.Command, args []string) { - log := logging.New(&logging.Config{}) + log := logging.New() ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/cmd/agent/daemon/conntrack/conntrack.go b/cmd/agent/daemon/conntrack/conntrack.go index 4bf98c157..99c63106d 100644 --- a/cmd/agent/daemon/conntrack/conntrack.go +++ b/cmd/agent/daemon/conntrack/conntrack.go @@ -4,7 +4,7 @@ import ( "net/netip" "os" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/florianl/go-conntrack" "github.com/vishvananda/netns" diff --git a/cmd/agent/daemon/conntrack/conntrack_cilium_linux.go b/cmd/agent/daemon/conntrack/conntrack_cilium_linux.go index 663e4f265..40a412c04 100644 --- a/cmd/agent/daemon/conntrack/conntrack_cilium_linux.go +++ b/cmd/agent/daemon/conntrack/conntrack_cilium_linux.go @@ -6,8 +6,8 @@ import ( "net/netip" "path/filepath" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" "github.com/cilium/cilium/pkg/bpf" "github.com/cilium/cilium/pkg/defaults" "github.com/cilium/cilium/pkg/loadbalancer" diff --git a/cmd/agent/daemon/conntrack/conntrack_cilium_other.go b/cmd/agent/daemon/conntrack/conntrack_cilium_other.go index 7f681e67f..067a1f718 100644 --- a/cmd/agent/daemon/conntrack/conntrack_cilium_other.go +++ b/cmd/agent/daemon/conntrack/conntrack_cilium_other.go @@ -5,7 +5,7 @@ package conntrack import ( "net/netip" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) func iniCiliumMaps(log *logging.Logger) bool { diff --git a/cmd/agent/daemon/conntrack/conntrack_nf.go b/cmd/agent/daemon/conntrack/conntrack_nf.go index cdc1ced54..1d37eeca0 100644 --- a/cmd/agent/daemon/conntrack/conntrack_nf.go +++ b/cmd/agent/daemon/conntrack/conntrack_nf.go @@ -6,7 +6,7 @@ import ( "strings" "syscall" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/florianl/go-conntrack" "github.com/samber/lo" ) diff --git a/cmd/agent/daemon/conntrack/conntrack_test.go b/cmd/agent/daemon/conntrack/conntrack_test.go deleted file mode 100644 index 792fe48f6..000000000 --- a/cmd/agent/daemon/conntrack/conntrack_test.go +++ /dev/null @@ -1,29 +0,0 @@ -package conntrack - -import ( - "fmt" - "log/slog" - "net/netip" - "testing" - - "github.com/castai/kvisor/pkg/logging" - "github.com/stretchr/testify/require" -) - -func TestConntrack(t *testing.T) { - t.Skip() // This test used for debug only. - - r := require.New(t) - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) - - ct, err := NewClient(log) - r.NoError(err) - src := netip.MustParseAddrPort("10.244.0.7:42861") - dsr := netip.MustParseAddrPort("10.244.0.65:8090") - res, found := ct.GetDestination(src, dsr) - _ = res - _ = found - fmt.Print("res", res, found) -} diff --git a/cmd/agent/daemon/debug/netflows.go b/cmd/agent/daemon/debug/netflows.go index db5a1cfd4..212e93df9 100644 --- a/cmd/agent/daemon/debug/netflows.go +++ b/cmd/agent/daemon/debug/netflows.go @@ -9,16 +9,17 @@ import ( "syscall" "time" + "github.com/jedib0t/go-pretty/v6/table" + "github.com/spf13/cobra" + "github.com/castai/kvisor/cmd/agent/daemon/cri" "github.com/castai/kvisor/pkg/cgroup" "github.com/castai/kvisor/pkg/containers" "github.com/castai/kvisor/pkg/ebpftracer" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" - "github.com/jedib0t/go-pretty/v6/table" - "github.com/spf13/cobra" + "github.com/castai/logging" ) func NewNetflowsDebugCommand() *cobra.Command { @@ -35,7 +36,7 @@ func NewNetflowsDebugCommand() *cobra.Command { limit := cmd.Flags().Int("limit", 500, "Limit netflows output") cmd.RunE = func(cmd *cobra.Command, args []string) error { - log := logging.New(&logging.Config{}) + log := logging.New() procHandler := proc.New() diff --git a/cmd/agent/daemon/debug/sockets.go b/cmd/agent/daemon/debug/sockets.go index ee3d96042..e2676802d 100644 --- a/cmd/agent/daemon/debug/sockets.go +++ b/cmd/agent/daemon/debug/sockets.go @@ -6,11 +6,12 @@ import ( "os" "strconv" - "github.com/castai/kvisor/pkg/ebpftracer/debug" - "github.com/castai/kvisor/pkg/logging" "github.com/jedib0t/go-pretty/v6/table" "github.com/spf13/cobra" "golang.org/x/sys/unix" + + "github.com/castai/kvisor/pkg/ebpftracer/debug" + "github.com/castai/logging" ) func NewSocketDebugCommand() *cobra.Command { @@ -23,7 +24,7 @@ func NewSocketDebugCommand() *cobra.Command { ) cmd.RunE = func(cmd *cobra.Command, args []string) error { - log := logging.New(&logging.Config{}) + log := logging.New() d, err := debug.New(log, debug.DebugCfg{ TargetPID: *targetPID, }) diff --git a/cmd/agent/daemon/enrichment/enrichers.go b/cmd/agent/daemon/enrichment/enrichers.go index 4ead291fa..c90d2723c 100644 --- a/cmd/agent/daemon/enrichment/enrichers.go +++ b/cmd/agent/daemon/enrichment/enrichers.go @@ -13,8 +13,8 @@ import ( "github.com/castai/kvisor/cmd/agent/daemon/metrics" "github.com/castai/kvisor/pkg/containers" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" "github.com/cespare/xxhash/v2" "github.com/elastic/go-freelru" "github.com/minio/sha256-simd" diff --git a/cmd/agent/daemon/enrichment/enrichers_test.go b/cmd/agent/daemon/enrichment/enrichers_test.go index 2633a26e4..81efc53c9 100644 --- a/cmd/agent/daemon/enrichment/enrichers_test.go +++ b/cmd/agent/daemon/enrichment/enrichers_test.go @@ -10,11 +10,12 @@ import ( "testing/fstest" "time" + "github.com/stretchr/testify/require" + castpb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" - "github.com/stretchr/testify/require" + "github.com/castai/logging" ) func TestFileHashEnricher(t *testing.T) { @@ -29,7 +30,7 @@ func TestFileHashEnricher(t *testing.T) { } enricher := EnrichWithFileHash( - logging.New(&logging.Config{}), + logging.New(), createDummyMntNSPIDStore(0, 10), fsys) @@ -82,7 +83,7 @@ func TestFileHashEnricher(t *testing.T) { fileName := "test" fsys := fstest.MapFS{} - enricher := EnrichWithFileHash(logging.New(&logging.Config{}), + enricher := EnrichWithFileHash(logging.New(), createDummyMntNSPIDStore(0, 1), fsys) @@ -111,7 +112,7 @@ func TestFileHashEnricher(t *testing.T) { r := require.New(t) fsys := fstest.MapFS{} - enricher := EnrichWithFileHash(logging.New(&logging.Config{}), + enricher := EnrichWithFileHash(logging.New(), createDummyMntNSPIDStore(0, 1), fsys) @@ -141,7 +142,7 @@ func TestFileHashEnricher(t *testing.T) { } enricher := EnrichWithFileHash( - logging.New(&logging.Config{}), + logging.New(), createDummyMntNSPIDStore(0, pid), fsys) @@ -202,7 +203,7 @@ func TestFileHashEnricher(t *testing.T) { } enricher := EnrichWithFileHash( - logging.New(&logging.Config{}), + logging.New(), createDummyMntNSPIDStore(mountNSID, pid), fsys) @@ -243,7 +244,7 @@ func TestFileHashEnricher(t *testing.T) { } enricher := EnrichWithFileHash( - logging.New(&logging.Config{}), + logging.New(), createDummyMntNSPIDStore(mountNSID, rootPid, pid), fsys) diff --git a/cmd/agent/daemon/enrichment/service.go b/cmd/agent/daemon/enrichment/service.go index 6a4b9a5eb..d2c41f09f 100644 --- a/cmd/agent/daemon/enrichment/service.go +++ b/cmd/agent/daemon/enrichment/service.go @@ -7,7 +7,7 @@ import ( castpb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/cmd/agent/daemon/metrics" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) type EnrichedContainerEvent struct { diff --git a/cmd/agent/daemon/enrichment/service_test.go b/cmd/agent/daemon/enrichment/service_test.go index 842199d6e..4632ca9d7 100644 --- a/cmd/agent/daemon/enrichment/service_test.go +++ b/cmd/agent/daemon/enrichment/service_test.go @@ -7,7 +7,7 @@ import ( castpb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) @@ -16,7 +16,7 @@ func TestEnrichmentService(t *testing.T) { r := require.New(t) var wantedPID uint32 = 99 - svc := NewService(logging.New(&logging.Config{}), Config{ + svc := NewService(logging.New(), Config{ WorkerCount: 1, EventEnrichers: []EventEnricher{ enricherFrom(func(ctx context.Context, req *EnrichedContainerEvent) { diff --git a/cmd/agent/daemon/export/castai/castai_databatch_writer.go b/cmd/agent/daemon/export/castai/castai_databatch_writer.go index 724396e33..368660870 100644 --- a/cmd/agent/daemon/export/castai/castai_databatch_writer.go +++ b/cmd/agent/daemon/export/castai/castai_databatch_writer.go @@ -7,7 +7,7 @@ import ( castaipb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/cmd/agent/daemon/export" "github.com/castai/kvisor/pkg/castai" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/cenkalti/backoff/v5" "google.golang.org/grpc" ) diff --git a/cmd/agent/daemon/export/clickhouse/clickhouse_process_tree_exporter.go b/cmd/agent/daemon/export/clickhouse/clickhouse_process_tree_exporter.go index f79b42d2e..a54ff3564 100644 --- a/cmd/agent/daemon/export/clickhouse/clickhouse_process_tree_exporter.go +++ b/cmd/agent/daemon/export/clickhouse/clickhouse_process_tree_exporter.go @@ -6,8 +6,8 @@ import ( "github.com/ClickHouse/clickhouse-go/v2" "github.com/castai/kvisor/cmd/agent/daemon/metrics" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/processtree" + "github.com/castai/logging" ) const ( diff --git a/cmd/agent/daemon/pipeline/cluster_info_pipeline.go b/cmd/agent/daemon/pipeline/cluster_info_pipeline.go index 55553d16b..c3777b560 100644 --- a/cmd/agent/daemon/pipeline/cluster_info_pipeline.go +++ b/cmd/agent/daemon/pipeline/cluster_info_pipeline.go @@ -8,7 +8,7 @@ import ( "time" kubepb "github.com/castai/kvisor/api/v1/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) const ( diff --git a/cmd/agent/daemon/pipeline/cluster_info_pipleline_test.go b/cmd/agent/daemon/pipeline/cluster_info_pipleline_test.go index 3e98f13a2..f9ef2eee6 100644 --- a/cmd/agent/daemon/pipeline/cluster_info_pipleline_test.go +++ b/cmd/agent/daemon/pipeline/cluster_info_pipleline_test.go @@ -6,15 +6,16 @@ import ( "testing" "time" - kubepb "github.com/castai/kvisor/api/v1/kube" - "github.com/castai/kvisor/pkg/logging" "github.com/stretchr/testify/require" "google.golang.org/grpc" + + kubepb "github.com/castai/kvisor/api/v1/kube" + "github.com/castai/logging" ) func TestNewClusterInfo(t *testing.T) { r := require.New(t) - log := logging.NewTestLog() + log := logging.New() mockClient := &mockKubeClient{} info := newClusterInfo(mockClient, log) @@ -74,7 +75,7 @@ func TestClusterInfoSync(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { r := require.New(t) - log := logging.NewTestLog() + log := logging.New() mockClient := &mockKubeClientWithRetry{ response: tt.response, diff --git a/cmd/agent/daemon/pipeline/controller.go b/cmd/agent/daemon/pipeline/controller.go index bae15a7bc..23df9d696 100644 --- a/cmd/agent/daemon/pipeline/controller.go +++ b/cmd/agent/daemon/pipeline/controller.go @@ -25,8 +25,8 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/signature" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/processtree" + "github.com/castai/logging" custommetrics "github.com/castai/metrics" ) diff --git a/cmd/agent/daemon/pipeline/controller_test.go b/cmd/agent/daemon/pipeline/controller_test.go index 953ad6ffb..227b91cdc 100644 --- a/cmd/agent/daemon/pipeline/controller_test.go +++ b/cmd/agent/daemon/pipeline/controller_test.go @@ -31,8 +31,8 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/signature" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/processtree" + "github.com/castai/logging" ) type testAddr struct { @@ -895,7 +895,7 @@ type customizeMockTracer func(t *mockEbpfTracer) type customizeMockContainersClient func(t *mockContainersClient) func newTestController(opts ...any) *Controller { - log := logging.NewTestLog() + log := logging.New() cfg := Config{ Stats: config.StatsConfig{ Enabled: false, diff --git a/cmd/agent/daemon/pipeline/storage_info_provider.go b/cmd/agent/daemon/pipeline/storage_info_provider.go index ac2bd7c79..ddeabf3ec 100644 --- a/cmd/agent/daemon/pipeline/storage_info_provider.go +++ b/cmd/agent/daemon/pipeline/storage_info_provider.go @@ -20,7 +20,7 @@ import ( "google.golang.org/grpc/encoding/gzip" kubepb "github.com/castai/kvisor/api/v1/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) const sectorSizeBytes = 512 diff --git a/cmd/agent/daemon/pipeline/storage_info_provider_test.go b/cmd/agent/daemon/pipeline/storage_info_provider_test.go index 44cb26e59..3523ee634 100644 --- a/cmd/agent/daemon/pipeline/storage_info_provider_test.go +++ b/cmd/agent/daemon/pipeline/storage_info_provider_test.go @@ -11,7 +11,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) // readProcDiskStatsFromPath is a test helper that reads diskstats from a custom path @@ -414,7 +414,7 @@ func TestCollectBlockDeviceMetrics(t *testing.T) { tmpFile2.Close() provider := &SysfsStorageInfoProvider{ - log: logging.NewTestLog(), + log: logging.New(), nodeName: "test-node", kubeClient: nil, // Not needed for this test storageState: &storageMetricsState{ @@ -515,7 +515,7 @@ E:SUBSYSTEM=block r.NoError(err) provider := &SysfsStorageInfoProvider{ - log: logging.NewTestLog(), + log: logging.New(), nodeName: "test-node", kubeClient: nil, // Not needed for this test sysBlockPrefix: tmpDir, @@ -620,7 +620,7 @@ func TestCalculateBlockDeviceRates(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { provider := &SysfsStorageInfoProvider{ - log: logging.NewTestLog(), + log: logging.New(), } current := tt.current @@ -685,7 +685,7 @@ func TestGetDiskType(t *testing.T) { require.NoError(t, err) provider := &SysfsStorageInfoProvider{ - log: logging.NewTestLog(), + log: logging.New(), sysBlockPrefix: tmpDir, storageState: &storageMetricsState{ blockDevices: make(map[string]*BlockDeviceMetric), diff --git a/cmd/agent/daemon/process/process.go b/cmd/agent/daemon/process/process.go index 380d6ff0b..840c27c10 100644 --- a/cmd/agent/daemon/process/process.go +++ b/cmd/agent/daemon/process/process.go @@ -14,7 +14,7 @@ import ( "sync/atomic" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) var ( diff --git a/cmd/agent/daemon/process/process_test.go b/cmd/agent/daemon/process/process_test.go index 45092e1a8..27ef78161 100644 --- a/cmd/agent/daemon/process/process_test.go +++ b/cmd/agent/daemon/process/process_test.go @@ -2,17 +2,15 @@ package process import ( "fmt" - "log/slog" "testing" - "github.com/castai/kvisor/pkg/logging" "github.com/stretchr/testify/require" + + "github.com/castai/logging" ) func TestFindContainerID(t *testing.T) { - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) + log := logging.New() client := NewClient(log, "./testdata") tests := []struct { diff --git a/cmd/controller/app/app.go b/cmd/controller/app/app.go index c0a0321d1..8ef99ae2f 100644 --- a/cmd/controller/app/app.go +++ b/cmd/controller/app/app.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "log/slog" "net" "net/http" "net/http/pprof" @@ -19,7 +18,6 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/samber/lo" "golang.org/x/sync/errgroup" - "golang.org/x/time/rate" "google.golang.org/grpc" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" @@ -34,7 +32,8 @@ import ( "github.com/castai/kvisor/pkg/blobscache" "github.com/castai/kvisor/pkg/castai" "github.com/castai/kvisor/pkg/cloudprovider" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" + "github.com/castai/logging/components" ) func New(cfg config.Config, clientset kubernetes.Interface) *App { @@ -54,17 +53,13 @@ func (a *App) Run(ctx context.Context) error { cfg := a.cfg clientset := a.kubeClient - var log *logging.Logger - logCfg := &logging.Config{ - AddSource: true, - Level: logging.MustParseLevel(cfg.LogLevel), - RateLimiter: logging.RateLimiterConfig{ - Limit: rate.Every(cfg.LogRateInterval), - Burst: cfg.LogRateBurst, - Inform: true, - }, + errg, ctx := errgroup.WithContext(ctx) + + logHandlers := []logging.Handler{ + logging.NewTextHandler(logging.DefaultTextHandlerConfig), } var castaiClient *castai.Client + var log *logging.Logger if a.cfg.CastaiEnv.Valid() { var err error castaiClient, err = castai.NewClient(fmt.Sprintf("kvisor-controller/%s", cfg.Version), cfg.CastaiEnv) @@ -72,24 +67,34 @@ func (a *App) Run(ctx context.Context) error { return fmt.Errorf("setting up castai api client: %w", err) } defer castaiClient.Close() - castaiLogsExporter := castai.NewLogsExporter(castaiClient) - go castaiLogsExporter.Run(ctx) //nolint:errcheck - if a.cfg.PromMetricsExportEnabled { - castaiMetricsExporter := castai.NewPromMetricsExporter(log, castaiLogsExporter, prometheus.DefaultGatherer, castai.PromMetricsExporterConfig{ - PodName: a.cfg.PodName, - ExportInterval: a.cfg.PromMetricsExportInterval, - }) - go castaiMetricsExporter.Run(ctx) //nolint:errcheck + logsApiClient, err := components.NewAPIClient(components.Config{ + APIBaseURL: a.cfg.CastaiEnv.APIURL, + APIKey: a.cfg.CastaiEnv.APIKey, + ClusterID: a.cfg.CastaiEnv.ClusterID, + Component: "kvisor-controller", + Version: a.cfg.Version, + }) + if err != nil { + return fmt.Errorf("creating logs api client: %w", err) } + batchLogsApiClient := components.NewBatchClient(logsApiClient) + errg.Go(func() error { + return batchLogsApiClient.Run(ctx) + }) + logsExportHandler := logging.NewExportHandler(batchLogsApiClient, logging.DefaultExportHandlerConfig) + logHandlers = append(logHandlers, logsExportHandler) + log = logging.New(logHandlers...) - logCfg.Export = logging.ExportConfig{ - ExportFunc: castaiLogsExporter.ExportFunc(), - MinLevel: slog.LevelInfo, - } - log = logging.New(logCfg) + castaiMetricsExporter := castai.NewPromMetricsExporter(log, batchLogsApiClient, prometheus.DefaultGatherer, castai.PromMetricsExporterConfig{ + PodName: a.cfg.PodName, + ExportInterval: a.cfg.PromMetricsExportInterval, + }) + errg.Go(func() error { + return castaiMetricsExporter.Run(ctx) + }) } else { - log = logging.New(logCfg) + log = logging.New(logHandlers...) } log.Infof("running kvisor-controller, cluster_id=%s, grpc_addr=%s, version=%s", cfg.CastaiEnv.ClusterID, cfg.CastaiEnv.APIGrpcAddr, cfg.Version) @@ -103,14 +108,13 @@ func (a *App) Run(ctx context.Context) error { kubeClient := kube.NewClient(log, cfg.PodName, cfg.PodNamespace, k8sVersion, clientset) kubeClient.RegisterHandlers(informersFactory) - errg, ctx := errgroup.WithContext(ctx) errg.Go(func() error { return kubeClient.Run(ctx) }) // Initialize cloud provider if enabled if cfg.CloudProviderConfig.IsAnyControllerEnabled() { - provider, err := cloudprovider.NewProvider(ctx, cfg.CloudProviderConfig.CloudProvider) + provider, err := cloudprovider.NewProvider(ctx, log, cfg.CloudProviderConfig.CloudProvider) if err == nil { log.Infof("cloud provider %s initialized successfully", provider.Type()) diff --git a/cmd/controller/controllers/castai_controller.go b/cmd/controller/controllers/castai_controller.go index a2faab0fe..389ed1278 100644 --- a/cmd/controller/controllers/castai_controller.go +++ b/cmd/controller/controllers/castai_controller.go @@ -10,7 +10,7 @@ import ( castaipb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/cmd/controller/kube" "github.com/castai/kvisor/pkg/castai" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "golang.org/x/sync/errgroup" ) diff --git a/cmd/controller/controllers/castai_controller_test.go b/cmd/controller/controllers/castai_controller_test.go index 83730a3ef..074be786f 100644 --- a/cmd/controller/controllers/castai_controller_test.go +++ b/cmd/controller/controllers/castai_controller_test.go @@ -3,23 +3,21 @@ package controllers import ( "context" "errors" - "log/slog" "testing" "time" - castaipb "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/castai" - "github.com/castai/kvisor/pkg/logging" "github.com/stretchr/testify/require" "google.golang.org/grpc" "k8s.io/client-go/kubernetes/fake" + + castaipb "github.com/castai/kvisor/api/v1/runtime" + "github.com/castai/kvisor/cmd/controller/kube" + "github.com/castai/kvisor/pkg/castai" + "github.com/castai/logging" ) func TestCastaiController(t *testing.T) { - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) + log := logging.New() k8sClient := fake.NewSimpleClientset() kubeClient := kube.NewClient(log, "agent", "ns", kube.Version{}, k8sClient) ctx, cancel := context.WithCancel(context.Background()) diff --git a/cmd/controller/controllers/imagescan/controller.go b/cmd/controller/controllers/imagescan/controller.go index c6166b280..34feb02a1 100644 --- a/cmd/controller/controllers/imagescan/controller.go +++ b/cmd/controller/controllers/imagescan/controller.go @@ -16,7 +16,7 @@ import ( corev1 "k8s.io/api/core/v1" "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/samber/lo" ) diff --git a/cmd/controller/controllers/imagescan/controller_test.go b/cmd/controller/controllers/imagescan/controller_test.go index 404890eb7..27d6c7f97 100644 --- a/cmd/controller/controllers/imagescan/controller_test.go +++ b/cmd/controller/controllers/imagescan/controller_test.go @@ -3,15 +3,11 @@ package imagescan import ( "context" "errors" - "log/slog" "sort" "sync" "testing" "time" - castaipb "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/logging" "github.com/google/uuid" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -21,14 +17,16 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + castaipb "github.com/castai/kvisor/api/v1/runtime" + "github.com/castai/kvisor/cmd/controller/kube" + "github.com/castai/logging" + imgcollectorconfig "github.com/castai/kvisor/cmd/imagescan/config" ) func TestSubscriber(t *testing.T) { ctx := context.Background() - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) + log := logging.New() createNode := func(name string) *corev1.Node { return &corev1.Node{ diff --git a/cmd/controller/controllers/jobs_cleanup_controller.go b/cmd/controller/controllers/jobs_cleanup_controller.go index 9d6def395..3d84b452d 100644 --- a/cmd/controller/controllers/jobs_cleanup_controller.go +++ b/cmd/controller/controllers/jobs_cleanup_controller.go @@ -6,7 +6,7 @@ import ( "fmt" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/samber/lo" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" diff --git a/cmd/controller/controllers/jobs_cleanup_controller_test.go b/cmd/controller/controllers/jobs_cleanup_controller_test.go index 72ed6f938..b5a22ccb3 100644 --- a/cmd/controller/controllers/jobs_cleanup_controller_test.go +++ b/cmd/controller/controllers/jobs_cleanup_controller_test.go @@ -6,7 +6,7 @@ import ( "testing" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" batchv1 "k8s.io/api/batch/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -17,7 +17,7 @@ func TestJobsCleanupController(t *testing.T) { r := require.New(t) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - log := logging.NewTestLog() + log := logging.New() ns := "castai-agent" oldJob := &batchv1.Job{ diff --git a/cmd/controller/controllers/kubebench/controller.go b/cmd/controller/controllers/kubebench/controller.go index 68ed1ab37..15b71e4ba 100644 --- a/cmd/controller/controllers/kubebench/controller.go +++ b/cmd/controller/controllers/kubebench/controller.go @@ -14,7 +14,7 @@ import ( castaipb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/cmd/controller/controllers/kubebench/spec" "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/cenkalti/backoff/v5" "github.com/cespare/xxhash/v2" lru "github.com/elastic/go-freelru" diff --git a/cmd/controller/controllers/kubebench/controller_test.go b/cmd/controller/controllers/kubebench/controller_test.go index 7ab69c3ad..0d2e13fc9 100644 --- a/cmd/controller/controllers/kubebench/controller_test.go +++ b/cmd/controller/controllers/kubebench/controller_test.go @@ -9,11 +9,10 @@ import ( "testing" "time" - castaipb "github.com/castai/kvisor/api/v1/runtime" "k8s.io/client-go/kubernetes" - "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/logging" + castaipb "github.com/castai/kvisor/api/v1/runtime" + "github.com/google/uuid" "github.com/stretchr/testify/require" "google.golang.org/grpc" @@ -21,6 +20,9 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes/fake" + + "github.com/castai/kvisor/cmd/controller/kube" + "github.com/castai/logging" ) var castaiNamespace = "castai-sec" @@ -31,7 +33,7 @@ func TestSubscriber(t *testing.T) { ctx := context.Background() clientset := fake.NewSimpleClientset() mockCast := &mockCastaiClient{} - log := logging.NewTestLog() + log := logging.New() logProvider := newMockLogProvider(readReport()) kubeCtrl := &mockKubeController{} @@ -92,7 +94,7 @@ func TestSubscriber(t *testing.T) { ctx := context.Background() clientset := fake.NewSimpleClientset() mockCast := &mockCastaiClient{} - log := logging.NewTestLog() + log := logging.New() logProvider := newMockLogProvider(readReport()) kubeCtrl := &mockKubeController{} @@ -133,7 +135,7 @@ func TestSubscriber(t *testing.T) { clientset := fake.NewSimpleClientset() mockCast := &mockCastaiClient{} - log := logging.NewTestLog() + log := logging.New() logProvider := newMockLogProvider(readReport()) kubeCtrl := &mockKubeController{} diff --git a/cmd/controller/controllers/kubelinter/controller.go b/cmd/controller/controllers/kubelinter/controller.go index 457e50a0d..da8f53e4f 100644 --- a/cmd/controller/controllers/kubelinter/controller.go +++ b/cmd/controller/controllers/kubelinter/controller.go @@ -10,7 +10,7 @@ import ( castaipb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "google.golang.org/grpc" appsv1 "k8s.io/api/apps/v1" batchv1 "k8s.io/api/batch/v1" diff --git a/cmd/controller/controllers/kubelinter/controller_test.go b/cmd/controller/controllers/kubelinter/controller_test.go index 7150c33e2..efcfcbab1 100644 --- a/cmd/controller/controllers/kubelinter/controller_test.go +++ b/cmd/controller/controllers/kubelinter/controller_test.go @@ -5,18 +5,19 @@ import ( rbacv1 "k8s.io/api/rbac/v1" "testing" - castaipb "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/cmd/controller/kube" - "github.com/castai/kvisor/pkg/logging" "github.com/samber/lo" "github.com/stretchr/testify/require" "google.golang.org/grpc" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + castaipb "github.com/castai/kvisor/api/v1/runtime" + "github.com/castai/kvisor/cmd/controller/kube" + "github.com/castai/logging" ) func TestSubscriber(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() t.Run("sends linter checks", func(t *testing.T) { r := require.New(t) diff --git a/cmd/controller/controllers/volume_state_controller.go b/cmd/controller/controllers/volume_state_controller.go index 28d79d0bc..6f338b569 100644 --- a/cmd/controller/controllers/volume_state_controller.go +++ b/cmd/controller/controllers/volume_state_controller.go @@ -7,7 +7,7 @@ import ( "github.com/castai/kvisor/cmd/controller/kube" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) type VolumeStateControllerConfig struct { diff --git a/cmd/controller/controllers/volume_state_controller_test.go b/cmd/controller/controllers/volume_state_controller_test.go index 11fdf01b4..48347a0b3 100644 --- a/cmd/controller/controllers/volume_state_controller_test.go +++ b/cmd/controller/controllers/volume_state_controller_test.go @@ -10,11 +10,11 @@ import ( "github.com/castai/kvisor/cmd/controller/kube" noop "github.com/castai/kvisor/pkg/cloudprovider/noop" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) func TestVolumeStateController(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() k8sClient := fake.NewSimpleClientset() kubeClient := kube.NewClient(log, "agent", "ns", kube.Version{}, k8sClient) diff --git a/cmd/controller/controllers/vpc_state_controller.go b/cmd/controller/controllers/vpc_state_controller.go index 27635b5ee..fae98466a 100644 --- a/cmd/controller/controllers/vpc_state_controller.go +++ b/cmd/controller/controllers/vpc_state_controller.go @@ -6,7 +6,7 @@ import ( "github.com/castai/kvisor/cmd/controller/kube" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) type VPCStateControllerConfig struct { diff --git a/cmd/controller/controllers/vpc_state_controller_test.go b/cmd/controller/controllers/vpc_state_controller_test.go index 4f29c3c32..553b3633a 100644 --- a/cmd/controller/controllers/vpc_state_controller_test.go +++ b/cmd/controller/controllers/vpc_state_controller_test.go @@ -8,13 +8,13 @@ import ( "github.com/castai/kvisor/cmd/controller/kube" noop "github.com/castai/kvisor/pkg/cloudprovider/noop" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" "k8s.io/client-go/kubernetes/fake" ) func TestVPCStateController(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() k8sClient := fake.NewSimpleClientset() kubeClient := kube.NewClient(log, "agent", "ns", kube.Version{}, k8sClient) diff --git a/cmd/controller/kube/client.go b/cmd/controller/kube/client.go index 152e09a68..fcc41231e 100644 --- a/cmd/controller/kube/client.go +++ b/cmd/controller/kube/client.go @@ -14,7 +14,7 @@ import ( "time" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" appsv1 "k8s.io/api/apps/v1" batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" diff --git a/cmd/controller/kube/client_test.go b/cmd/controller/kube/client_test.go index 0b6da4d38..2062e3def 100644 --- a/cmd/controller/kube/client_test.go +++ b/cmd/controller/kube/client_test.go @@ -9,7 +9,7 @@ import ( "testing" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/samber/lo" "github.com/stretchr/testify/require" appsv1 "k8s.io/api/apps/v1" @@ -22,7 +22,7 @@ import ( func TestClient(t *testing.T) { ctx := context.Background() - log := logging.NewTestLog() + log := logging.New() clientset := fake.NewClientset() listener := &mockListener{items: map[types.UID]Object{}} diff --git a/cmd/controller/kube/index_volume.go b/cmd/controller/kube/index_volume.go index 7cd924c57..359b8d22b 100644 --- a/cmd/controller/kube/index_volume.go +++ b/cmd/controller/kube/index_volume.go @@ -4,7 +4,7 @@ import ( "sync" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) // VolumeIndex maintains a mapping of node names to cloud volumes. diff --git a/cmd/controller/kube/index_volume_test.go b/cmd/controller/kube/index_volume_test.go index 09f37bd38..4cbf163d1 100644 --- a/cmd/controller/kube/index_volume_test.go +++ b/cmd/controller/kube/index_volume_test.go @@ -4,12 +4,12 @@ import ( "testing" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) func TestVolumeIndex(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() t.Run("new volume index", func(t *testing.T) { r := require.New(t) diff --git a/cmd/controller/kube/index_vpc.go b/cmd/controller/kube/index_vpc.go index f79a47ef5..71320822f 100644 --- a/cmd/controller/kube/index_vpc.go +++ b/cmd/controller/kube/index_vpc.go @@ -7,7 +7,7 @@ import ( "github.com/castai/kvisor/pkg/cidrindex" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) // IPVPCInfo contains network state for a specific IP address. diff --git a/cmd/controller/kube/index_vpc_test.go b/cmd/controller/kube/index_vpc_test.go index 1772f2dea..7af77c1e8 100644 --- a/cmd/controller/kube/index_vpc_test.go +++ b/cmd/controller/kube/index_vpc_test.go @@ -6,12 +6,12 @@ import ( "time" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) func TestVPCIndex(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() t.Run("new VPC index", func(t *testing.T) { r := require.New(t) diff --git a/cmd/controller/kube/node_stats_test.go b/cmd/controller/kube/node_stats_test.go index c328a6631..e55114ad0 100644 --- a/cmd/controller/kube/node_stats_test.go +++ b/cmd/controller/kube/node_stats_test.go @@ -7,7 +7,7 @@ import ( "testing" kubepb "github.com/castai/kvisor/api/v1/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -15,7 +15,7 @@ import ( ) func TestGetNodeStatsSummary(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() // Create mock stats summary response mockStats := NodeStatsSummary{ @@ -55,7 +55,7 @@ func TestGetNodeStatsSummary(t *testing.T) { }, } - _ = json.Marshal // Keep json import for potential future use + _ = json.Marshal // Keep json import for potential future use _ = http.StatusOK // Keep http import for potential future use _ = log // Keep log for consistency @@ -140,7 +140,7 @@ func TestGetNodeStatsSummary(t *testing.T) { } func TestServerGetNodeStatsSummary(t *testing.T) { - log := logging.NewTestLog() + log := logging.New() t.Run("empty node name returns error", func(t *testing.T) { clientset := fake.NewClientset() diff --git a/cmd/controller/kube/server_test.go b/cmd/controller/kube/server_test.go index 3afc6b869..0b29b9daa 100644 --- a/cmd/controller/kube/server_test.go +++ b/cmd/controller/kube/server_test.go @@ -7,7 +7,7 @@ import ( "testing" kubepb "github.com/castai/kvisor/api/v1/kube" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -21,7 +21,7 @@ import ( func TestServer(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - log := logging.NewTestLog() + log := logging.New() clientset := fake.NewClientset() client := NewClient(log, "castai-kvisor", "kvisor", Version{}, clientset) diff --git a/cmd/controller/main.go b/cmd/controller/main.go index be0923048..f6dca9c6c 100644 --- a/cmd/controller/main.go +++ b/cmd/controller/main.go @@ -10,17 +10,18 @@ import ( "syscall" "time" + "github.com/spf13/pflag" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + "k8s.io/client-go/util/flowcontrol" + "github.com/castai/kvisor/cmd/controller/config" "github.com/castai/kvisor/cmd/controller/controllers/imagescan" "github.com/castai/kvisor/cmd/controller/controllers/kubebench" "github.com/castai/kvisor/cmd/controller/controllers/kubelinter" "github.com/castai/kvisor/pkg/castai" cloudtypes "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/spf13/pflag" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/rest" - "k8s.io/client-go/tools/clientcmd" - "k8s.io/client-go/util/flowcontrol" "github.com/castai/kvisor/cmd/controller/app" "github.com/castai/kvisor/cmd/controller/controllers" diff --git a/cmd/event-generator/app/app.go b/cmd/event-generator/app/app.go index af4cc9b62..b5cf644fe 100644 --- a/cmd/event-generator/app/app.go +++ b/cmd/event-generator/app/app.go @@ -8,7 +8,7 @@ import ( "syscall" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" diff --git a/cmd/event-generator/app/controller.go b/cmd/event-generator/app/controller.go index 879b36ada..81b92d9eb 100644 --- a/cmd/event-generator/app/controller.go +++ b/cmd/event-generator/app/controller.go @@ -6,7 +6,7 @@ import ( "math/rand" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/samber/lo" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" diff --git a/cmd/event-generator/app/event.go b/cmd/event-generator/app/event.go index e637aff92..baba170e7 100644 --- a/cmd/event-generator/app/event.go +++ b/cmd/event-generator/app/event.go @@ -10,7 +10,7 @@ import ( "strings" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) var ( diff --git a/cmd/event-generator/app/thief.go b/cmd/event-generator/app/thief.go index c6f0c5cb2..9382604a7 100644 --- a/cmd/event-generator/app/thief.go +++ b/cmd/event-generator/app/thief.go @@ -6,7 +6,7 @@ import ( "strings" "time" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/redis/go-redis/v9" ) diff --git a/cmd/event-generator/main.go b/cmd/event-generator/main.go index c0ed19a92..14082919d 100644 --- a/cmd/event-generator/main.go +++ b/cmd/event-generator/main.go @@ -3,10 +3,11 @@ package main import ( "time" - "github.com/castai/kvisor/pkg/logging" "github.com/sirupsen/logrus" "github.com/spf13/pflag" + "github.com/castai/logging" + "github.com/castai/kvisor/cmd/event-generator/app" ) @@ -21,9 +22,9 @@ var ( func main() { pflag.Parse() - log := logging.New(&logging.Config{ + log := logging.New(logging.NewTextHandler(logging.TextHandlerConfig{ Level: logging.MustParseLevel(*logLevel), - }) + })) genapp, err := app.New(&app.Config{ Version: Version, diff --git a/cmd/mock-server/main.go b/cmd/mock-server/main.go index 1dfd9397d..83f0eeefc 100644 --- a/cmd/mock-server/main.go +++ b/cmd/mock-server/main.go @@ -1,45 +1,64 @@ package main import ( + "compress/gzip" "context" + "encoding/json" "fmt" + "golang.org/x/sync/errgroup" + "io" "log/slog" "net" + "net/http" "os/signal" + "strings" "syscall" - castaipb "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/pkg/logging" "google.golang.org/grpc" _ "google.golang.org/grpc/encoding/gzip" + + castaipb "github.com/castai/kvisor/api/v1/runtime" + "github.com/castai/logging" + "github.com/castai/logging/components" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer cancel() - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) + cfg := logging.DefaultTextHandlerConfig + cfg.Level = slog.LevelDebug + log := logging.New(logging.NewTextHandler(cfg)) // nolint:gosec - lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", ":8443") + grpcLis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", ":8443") if err != nil { log.Fatal(err.Error()) } - defer lis.Close() + defer grpcLis.Close() - srv := grpc.NewServer() - castaipb.RegisterRuntimeSecurityAgentAPIServer(srv, NewMockServer(log)) + grpcServer := grpc.NewServer() + castaipb.RegisterRuntimeSecurityAgentAPIServer(grpcServer, NewMockServer(log)) go func() { <-ctx.Done() log.Info("shutting down grpc ingestor server") - srv.Stop() + grpcServer.Stop() }() - fmt.Println("listening at :8443") - if err := srv.Serve(lis); err != nil { + var errg errgroup.Group + errg.Go(func() error { + fmt.Println("listening grpc at :8443") + return grpcServer.Serve(grpcLis) + }) + + errg.Go(func() error { + log.Info("listening http at :8080") + httpServer := &testCASTAIHTTPServer{} + return http.ListenAndServe(":8080", httpServer) //nolint:gosec + }) + + if err := errg.Wait(); err != nil { log.Fatal(err.Error()) } } @@ -102,3 +121,32 @@ func (m *MockServer) LogsWriteStream(server castaipb.RuntimeSecurityAgentAPI_Log m.log.Debugf("log: %v", event) } } + +type testCASTAIHTTPServer struct { +} + +func (s *testCASTAIHTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + var bodyReader io.Reader = r.Body + if r.Header.Get("Content-Encoding") == "gzip" { + gz, err := gzip.NewReader(r.Body) + if err != nil { + http.Error(w, "invalid gzip body", http.StatusBadRequest) + return + } + defer gz.Close() + bodyReader = gz + } + + if strings.HasSuffix(r.URL.Path, "/logs") { + var req components.IngestLogsRequest + if err := json.NewDecoder(bodyReader).Decode(&req); err != nil { + fmt.Printf("failed to decode logs request: %v\n", err) + w.WriteHeader(http.StatusBadRequest) + return + } + fmt.Printf("received logs\n") + for _, entry := range req.Entries { + fmt.Printf(" level=%v msg=%v\n", entry.Level, entry.Message) + } + } +} diff --git a/e2e/e2e.go b/e2e/e2e.go index b48f0689f..c4205e414 100644 --- a/e2e/e2e.go +++ b/e2e/e2e.go @@ -2,12 +2,14 @@ package main import ( "bytes" + "compress/gzip" "context" "encoding/json" "errors" "fmt" "io" "net" + "net/http" "net/netip" "os" "os/exec" @@ -30,6 +32,7 @@ import ( castaipb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/cmd/agent/daemon/pipeline" + "github.com/castai/logging/components" metricspb "github.com/castai/metrics/api/v1beta" ) @@ -71,12 +74,21 @@ func run(ctx context.Context) error { return errors.New("image-tag flag is not set") } - addr := fmt.Sprintf(":%d", 8443) - lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", addr) + httpAddr := fmt.Sprintf(":%d", 8080) + httpServer := &testCASTAIHTTPServer{} + go func() { + //nolint:gosec + if err := http.ListenAndServe(httpAddr, httpServer); err != nil && !errors.Is(err, http.ErrServerClosed) { + fmt.Printf("http server listen error: %v\n", err) + } + }() + + grpcAddr := fmt.Sprintf(":%d", 8443) + grpcListener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", grpcAddr) if err != nil { return err } - s := grpc.NewServer() + grpcServer := grpc.NewServer() inClusterConfig, err := rest.InClusterConfig() if err != nil { return err @@ -87,10 +99,10 @@ func run(ctx context.Context) error { } srv := &testCASTAIServer{clientset: clientset, testStartTime: time.Now().UTC(), outputReceivedData: false} - castaipb.RegisterRuntimeSecurityAgentAPIServer(s, srv) - metricspb.RegisterIngestionAPIServer(s, srv) + castaipb.RegisterRuntimeSecurityAgentAPIServer(grpcServer, srv) + metricspb.RegisterIngestionAPIServer(grpcServer, srv) go func() { - if err := s.Serve(lis); err != nil { + if err := grpcServer.Serve(grpcListener); err != nil { fmt.Printf("serving grcp failed: %v\n", err) } }() @@ -180,11 +192,15 @@ func run(ctx context.Context) error { srv.storageMetricsAsserted = true srv.storageMetrics = nil - fmt.Println("🙏waiting for flogs") - if err := srv.assertLogs(ctx); err != nil { - return fmt.Errorf("assert logs: %w", err) + fmt.Println("🙏waiting for kvisor agent logs") + if err := httpServer.assertLogs(ctx, "kvisor-agent"); err != nil { + return fmt.Errorf("assert kvisor logs: %w", err) + } + + fmt.Println("🙏waiting for kvisor controller logs") + if err := httpServer.assertLogs(ctx, "kvisor-controller"); err != nil { + return fmt.Errorf("assert controller logs: %w", err) } - srv.logs = nil fmt.Println("👌e2e finished") @@ -199,6 +215,7 @@ func installChart(ctx context.Context, ns, imageTag string) ([]byte, error) { } grpcAddr := fmt.Sprintf("%s:8443", os.Getenv("POD_IP")) + httpAddr := fmt.Sprintf("http://%s:8080", os.Getenv("POD_IP")) //nolint:gosec cmd := exec.CommandContext(ctx, "/bin/sh", "-c", fmt.Sprintf( `helm upgrade --install kvisor-e2e ./charts/kvisor \ @@ -236,6 +253,7 @@ func installChart(ctx context.Context, ns, imageTag string) ([]byte, error) { --set castai.grpcAddr=%s \ --set castai.apiKey=%s \ --set castai.clusterID=%s \ + --set castai.apiURL=%s \ --wait --timeout=5m`, ns, repo, @@ -243,6 +261,7 @@ func installChart(ctx context.Context, ns, imageTag string) ([]byte, error) { grpcAddr, apiKey, clusterID, + httpAddr, )) return cmd.CombinedOutput() } @@ -261,7 +280,6 @@ type testCASTAIServer struct { containerEvents []*castaipb.ContainerEvents eventsAsserted bool - logs []*castaipb.LogEvent imageMetadatas []*castaipb.ImageMetadata kubeBenchReports []*castaipb.KubeBenchReport kubeLinterReports []*castaipb.KubeLinterReport @@ -515,50 +533,7 @@ func (t *testCASTAIServer) GetConfiguration(ctx context.Context, req *castaipb.G } func (t *testCASTAIServer) LogsWriteStream(server castaipb.RuntimeSecurityAgentAPI_LogsWriteStreamServer) error { - for { - event, err := server.Recv() - if err != nil { - return err - } - if t.outputReceivedData { - fmt.Println("received log:", event) - } - t.mu.Lock() - t.logs = append(t.logs, event) - t.mu.Unlock() - } -} - -func (t *testCASTAIServer) assertLogs(ctx context.Context) error { - timeout := time.After(10 * time.Second) - - r := newAssertions() - - for { - select { - case <-ctx.Done(): - return ctx.Err() - case <-timeout: - return errors.New("timeout waiting for received logs") - case <-time.After(1 * time.Second): - t.mu.Lock() - logs := slices.Clone(t.logs) - t.mu.Unlock() - - if len(logs) > 0 { - l1 := logs[0] - r.NotEmpty(l1.Msg) - - for _, l := range logs { - if strings.Contains(l.Msg, "panic") { - return fmt.Errorf("received logs contains panic error: %s", l.Msg) - } - } - - return r.error() - } - } - } + return errors.New("should not be called") } func (t *testCASTAIServer) assertStorageMetrics(ctx context.Context) error { @@ -1523,3 +1498,88 @@ func newAssertions() *assertions { t: t, } } + +type testCASTAIHTTPServer struct { + mu sync.Mutex + agentLogs []components.Entry + controllerLogs []components.Entry +} + +func (s *testCASTAIHTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + fmt.Printf("HTTP req: %v\n", r.URL.Path) + switch r.URL.Path { + case fmt.Sprintf("/v1/clusters/%s/components/kvisor-agent/logs", clusterID): + s.handleLogs("kvisor-agent")(w, r) + case fmt.Sprintf("/v1/clusters/%s/components/kvisor-controller/logs", clusterID): + s.handleLogs("kvisor-controller")(w, r) + } +} + +func (s *testCASTAIHTTPServer) handleLogs(component string) func(http.ResponseWriter, *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + var bodyReader io.Reader = r.Body + if r.Header.Get("Content-Encoding") == "gzip" { + gz, err := gzip.NewReader(r.Body) + if err != nil { + http.Error(w, "invalid gzip body", http.StatusBadRequest) + return + } + defer gz.Close() + bodyReader = gz + } + + var req components.IngestLogsRequest + if err := json.NewDecoder(bodyReader).Decode(&req); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + s.mu.Lock() + defer s.mu.Unlock() + + switch component { + case "kvisor-agent": + s.agentLogs = append(s.agentLogs, req.Entries...) + case "kvisor-controller": + s.controllerLogs = append(s.controllerLogs, req.Entries...) + } + w.WriteHeader(http.StatusOK) + } +} + +func (t *testCASTAIHTTPServer) assertLogs(ctx context.Context, component string) error { + timeout := time.After(10 * time.Second) + + r := newAssertions() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-timeout: + return errors.New("timeout waiting for received logs") + case <-time.After(1 * time.Second): + var logs []components.Entry + t.mu.Lock() + switch component { + case "kvisor-agent": + logs = slices.Clone(t.agentLogs) + case "kvisor-controller": + logs = slices.Clone(t.controllerLogs) + } + t.mu.Unlock() + + if len(logs) > 0 { + l1 := logs[0] + r.NotEmpty(l1.Message) + + for _, l := range logs { + if strings.Contains(l.Message, "panic") { + return fmt.Errorf("received logs contains panic error: %s", l.Message) + } + } + + return r.error() + } + } + } +} diff --git a/go.mod b/go.mod index 1c3f9e5ef..4318129a1 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/aws/aws-sdk-go-v2/config v1.29.13 github.com/aws/aws-sdk-go-v2/service/ec2 v1.279.1 github.com/castai/image-analyzer v0.10.0 - github.com/castai/logging v0.2.0 + github.com/castai/logging v0.3.0 github.com/castai/metrics v0.0.0-20250917084341-1533777a055a github.com/cenkalti/backoff/v5 v5.0.2 github.com/cespare/xxhash/v2 v2.3.0 @@ -54,7 +54,7 @@ require ( go.uber.org/atomic v1.11.0 go.uber.org/goleak v1.3.0 golang.org/x/net v0.46.0 - golang.org/x/sync v0.18.0 + golang.org/x/sync v0.19.0 golang.org/x/sys v0.37.0 golang.org/x/time v0.14.0 golang.stackrox.io/kube-linter v0.7.3-0.20250507172404-3f4b9037f56f diff --git a/go.sum b/go.sum index b4ffd076e..7408050f1 100644 --- a/go.sum +++ b/go.sum @@ -844,8 +844,10 @@ github.com/bytecodealliance/wasmtime-go/v3 v3.0.2 h1:3uZCA/BLTIu+DqCfguByNMJa2HV github.com/bytecodealliance/wasmtime-go/v3 v3.0.2/go.mod h1:RnUjnIXxEJcL6BgCvNyzCCRzZcxCgsZCi+RNlvYor5Q= github.com/castai/image-analyzer v0.10.0 h1:qJjJtbgsZHQtDci8FacbQ1v8xOk9nzWvnMR6j9x6MFg= github.com/castai/image-analyzer v0.10.0/go.mod h1:2ncQ2nejVVEqYPFdwZvq3Sl6G3DjM0ZIKS5copHQCxA= -github.com/castai/logging v0.2.0 h1:Zh0z1MWa3Baz3dd51N0xvPJKxvxoepPA5GV7kmNa8OI= -github.com/castai/logging v0.2.0/go.mod h1:Jm1Ad/+Ed8E+Fbhxw505Iu+3liYeOI8J55TBm3LnfRQ= +github.com/castai/logging v0.2.1-0.20260209131414-bd8d736fddb5 h1:ts1mTdTf/hOQPNTUjlpePsgupdVRDvrrGTm7Mc6B3xo= +github.com/castai/logging v0.2.1-0.20260209131414-bd8d736fddb5/go.mod h1:khOuV7bUcqeYx/4FSpJ1ntbe+JCHLaoR/PkY2Yc98rg= +github.com/castai/logging v0.3.0 h1:3S34to2Vn+vDrheeoi1GG2yVSjbZ58H3pvotGhq9Edg= +github.com/castai/logging v0.3.0/go.mod h1:khOuV7bUcqeYx/4FSpJ1ntbe+JCHLaoR/PkY2Yc98rg= github.com/castai/metrics v0.0.0-20250917084341-1533777a055a h1:wFcMa6dOMaMf3dGK9PIicZnRqBPIBxolvibHyA4k3to= github.com/castai/metrics v0.0.0-20250917084341-1533777a055a/go.mod h1:m63MWJd4H5a8Z21sGc0Ya0zci/JrSpR+/y+RCA1y59I= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= @@ -2203,8 +2205,8 @@ golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= -golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= -golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= diff --git a/pkg/blobscache/server.go b/pkg/blobscache/server.go index 95d278205..0dcbca728 100644 --- a/pkg/blobscache/server.go +++ b/pkg/blobscache/server.go @@ -4,7 +4,7 @@ import ( "fmt" "net/http" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/cespare/xxhash/v2" lru "github.com/elastic/go-freelru" "github.com/labstack/echo/v4" diff --git a/pkg/blobscache/server_test.go b/pkg/blobscache/server_test.go index dbca1c13f..8b556ad1d 100644 --- a/pkg/blobscache/server_test.go +++ b/pkg/blobscache/server_test.go @@ -4,15 +4,15 @@ import ( "context" "errors" "fmt" - "log/slog" "net/http/httptest" "testing" "time" - ia "github.com/castai/image-analyzer" - "github.com/castai/kvisor/pkg/logging" "github.com/labstack/echo/v4" "github.com/stretchr/testify/require" + + ia "github.com/castai/image-analyzer" + "github.com/castai/logging" ) func TestBlobsCacheServer(t *testing.T) { @@ -20,7 +20,7 @@ func TestBlobsCacheServer(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - log := logging.New(&logging.Config{Level: slog.LevelDebug}) + log := logging.New() srv := NewServer(log) e := echo.New() diff --git a/pkg/castai/client.go b/pkg/castai/client.go index 2480a2191..b4d062759 100644 --- a/pkg/castai/client.go +++ b/pkg/castai/client.go @@ -5,12 +5,13 @@ import ( "fmt" "strings" - castaipb "github.com/castai/kvisor/api/v1/runtime" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/encoding/gzip" "google.golang.org/grpc/metadata" + + castaipb "github.com/castai/kvisor/api/v1/runtime" ) func NewClient(userAgent string, cfg Config) (*Client, error) { diff --git a/pkg/castai/client_test.go b/pkg/castai/client_test.go index 013d5aff6..2822a67aa 100644 --- a/pkg/castai/client_test.go +++ b/pkg/castai/client_test.go @@ -9,13 +9,14 @@ import ( "testing" "time" - castaipb "github.com/castai/kvisor/api/v1/runtime" "github.com/google/uuid" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" "google.golang.org/grpc" "google.golang.org/grpc/encoding/gzip" "google.golang.org/grpc/metadata" + + castaipb "github.com/castai/kvisor/api/v1/runtime" ) func TestClient(t *testing.T) { @@ -99,6 +100,45 @@ func TestRemote(t *testing.T) { r.NoError(errg.Wait()) } +func TestGetAPIURL(t *testing.T) { + tests := []struct { + addr string + expected string + }{ + { + addr: "kvisor.dev-master.cast.ai:443", + expected: "https://api.dev-master.cast.ai", + }, + { + addr: "kvisor.prod-master.cast.ai:443", + expected: "https://api.cast.ai", + }, + { + addr: "kvisor.prod-eu.cast.ai:443", + expected: "https://api.eu.cast.ai", + }, + { + addr: "grpc-some-other-env.cast.ai:443", + expected: "https://api-some-other-env.cast.ai:443", + }, + { + addr: "api-grpc-some-other-env.cast.ai:443", + expected: "https://api-some-other-env.cast.ai:443", + }, + { + addr: "no-match-addr:443", + expected: "", + }, + } + + for _, test := range tests { + t.Run(test.addr, func(t *testing.T) { + r := require.New(t) + r.Equal(test.expected, getAPIURL(test.addr)) + }) + } +} + type testServer struct { logsWriteStreamHandler func(server castaipb.RuntimeSecurityAgentAPI_LogsWriteStreamServer) error } diff --git a/pkg/castai/config.go b/pkg/castai/config.go index 75b667669..84462d28c 100644 --- a/pkg/castai/config.go +++ b/pkg/castai/config.go @@ -3,11 +3,13 @@ package castai import ( "fmt" "os" + "strings" ) type Config struct { ClusterID string `json:"clusterID"` APIKey string `json:"-"` + APIURL string `json:"APIURL"` APIGrpcAddr string `json:"APIGrpcAddr"` Insecure bool `json:"insecure"` @@ -30,12 +32,21 @@ func NewConfigFromEnv(insecure bool) (Config, error) { return Config{}, err } - return Config{ + cfg := Config{ APIKey: apiKey, APIGrpcAddr: gRPCAddress, ClusterID: clusterID, Insecure: insecure, - }, nil + } + + apiURL, _ := lookupConfigVariable("API_URL") + if apiURL != "" { + cfg.APIURL = apiURL + } else { + cfg.APIURL = getAPIURL(gRPCAddress) + } + + return cfg, nil } func (c Config) Valid() bool { @@ -55,3 +66,25 @@ func lookupConfigVariable(name string) (string, error) { return "", fmt.Errorf("environment variable missing: please provide either `CAST_%s` or `%s`", name, name) } + +func getAPIURL(grpcAddr string) string { + envsMapping := map[string]string{ + "kvisor.dev-master.cast.ai:443": "https://api.dev-master.cast.ai", + "kvisor.prod-master.cast.ai:443": "https://api.cast.ai", + "kvisor.prod-eu.cast.ai:443": "https://api.eu.cast.ai", + } + for k, v := range envsMapping { + if grpcAddr == k { + return v + } + } + + // Other ennvs. + if strings.HasPrefix(grpcAddr, "api-grpc") { + return strings.Replace(grpcAddr, "api-grpc", "https://api", 1) + } + if strings.HasPrefix(grpcAddr, "grpc") { + return strings.Replace(grpcAddr, "grpc", "https://api", 1) + } + return "" +} diff --git a/pkg/castai/logs_exporter.go b/pkg/castai/logs_exporter.go deleted file mode 100644 index 1259d8165..000000000 --- a/pkg/castai/logs_exporter.go +++ /dev/null @@ -1,59 +0,0 @@ -package castai - -import ( - "context" - "log/slog" - "time" - - castaipb "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/pkg/logging" - "google.golang.org/grpc" -) - -func NewLogsExporter(client *Client) *LogsExporter { - return &LogsExporter{ - client: client, - logsChan: make(chan *castaipb.LogEvent, 1000), - } -} - -type LogsExporter struct { - client *Client - logsChan chan *castaipb.LogEvent -} - -func (l *LogsExporter) Run(ctx context.Context) error { - ws := NewWriteStream[*castaipb.LogEvent, *castaipb.SendLogsResponse](ctx, func(ctx context.Context) (grpc.ClientStream, error) { - return l.client.GRPC.LogsWriteStream(ctx) - }) - defer ws.Close() - ws.ReopenDelay = 1 * time.Second - - for { - select { - case <-ctx.Done(): - return ctx.Err() - case e := <-l.logsChan: - _ = ws.Send(e) - } - } -} - -func (l *LogsExporter) ExportFunc() logging.ExportFunc { - return func(ctx context.Context, record slog.Record) { - select { - case l.logsChan <- &castaipb.LogEvent{ - Level: int32(record.Level), // nolint:gosec - Msg: record.Message, - }: - default: - } - } -} - -func (l *LogsExporter) AddLogEvent(e *castaipb.LogEvent) { - select { - case l.logsChan <- e: - default: - } -} diff --git a/pkg/castai/prommetrics_exporter.go b/pkg/castai/prommetrics_exporter.go index 767b569c1..86fcee598 100644 --- a/pkg/castai/prommetrics_exporter.go +++ b/pkg/castai/prommetrics_exporter.go @@ -3,17 +3,17 @@ package castai import ( "context" "fmt" - "log/slog" "strings" "time" - castaipb "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" + "github.com/castai/logging/components" + "github.com/prometheus/client_golang/prometheus" ) type logsExporter interface { - AddLogEvent(e *castaipb.LogEvent) + IngestLogs(ctx context.Context, entries []components.Entry) error } type PromMetricsExporterConfig struct { @@ -81,9 +81,15 @@ func (e *PromMetricsExporter) export() error { msgs = append(msgs, fmt.Sprintf("%s %s %d", name, strings.Join(labels, " "), int(val))) } } - e.logsExporter.AddLogEvent(&castaipb.LogEvent{ - Level: int32(slog.LevelInfo), - Msg: fmt.Sprintf("kvisor metrics, pod=%s:\n%s", e.cfg.PodName, strings.Join(msgs, "\n")), + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return e.logsExporter.IngestLogs(ctx, []components.Entry{ + { + Level: string(components.LogLevelInfo), + Message: fmt.Sprintf("kvisor metrics, pod=%s:\n%s", e.cfg.PodName, strings.Join(msgs, "\n")), + Time: time.Now().UTC(), + Fields: nil, + }, }) - return nil } diff --git a/pkg/castai/prommetrics_exporter_test.go b/pkg/castai/prommetrics_exporter_test.go index 85b2429ad..00fdf2d6a 100644 --- a/pkg/castai/prommetrics_exporter_test.go +++ b/pkg/castai/prommetrics_exporter_test.go @@ -6,8 +6,9 @@ import ( "testing" "time" - "github.com/castai/kvisor/api/v1/runtime" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" + "github.com/castai/logging/components" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "github.com/stretchr/testify/require" @@ -32,7 +33,7 @@ func TestPrometheusExporter(t *testing.T) { }, []string{"event_type"}) testMetric3.WithLabelValues("other").Add(15) - log := logging.NewTestLog() + log := logging.New() logsExp := &mockLogsExporter{} exp := NewPromMetricsExporter(log, logsExp, prometheus.DefaultGatherer, PromMetricsExporterConfig{ MetricsPrefix: "kvisor_test", @@ -55,17 +56,18 @@ func TestPrometheusExporter(t *testing.T) { r.Len(logsExp.logs, 2) r.Equal(`kvisor metrics, pod=pod1: kvisor_test_counter event_type=exec 10 -kvisor_test_gauge event_name=Connect event_type=tcp_connect 15`, logsExp.logs[0].Msg) +kvisor_test_gauge event_name=Connect event_type=tcp_connect 15`, logsExp.logs[0].Message) } type mockLogsExporter struct { - logs []*v1.LogEvent + logs []components.Entry mu sync.Mutex } -func (m *mockLogsExporter) AddLogEvent(e *v1.LogEvent) { +func (m *mockLogsExporter) IngestLogs(ctx context.Context, entries []components.Entry) error { m.mu.Lock() defer m.mu.Unlock() - m.logs = append(m.logs, e) + m.logs = append(m.logs, entries...) + return nil } diff --git a/pkg/cgroup/client.go b/pkg/cgroup/client.go index db3b1925a..973c3e189 100644 --- a/pkg/cgroup/client.go +++ b/pkg/cgroup/client.go @@ -14,7 +14,7 @@ import ( "syscall" "github.com/castai/kvisor/cmd/agent/daemon/metrics" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) var ( diff --git a/pkg/cloudprovider/aws/auth.go b/pkg/cloudprovider/aws/auth.go index 4204dd306..4e8fdd32e 100644 --- a/pkg/cloudprovider/aws/auth.go +++ b/pkg/cloudprovider/aws/auth.go @@ -7,6 +7,7 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/config" + "github.com/castai/kvisor/pkg/cloudprovider/types" ) diff --git a/pkg/cloudprovider/aws/provider.go b/pkg/cloudprovider/aws/provider.go index da5fb8ffd..5e5fa2f88 100644 --- a/pkg/cloudprovider/aws/provider.go +++ b/pkg/cloudprovider/aws/provider.go @@ -6,8 +6,9 @@ import ( "sync" "github.com/aws/aws-sdk-go-v2/service/ec2" + "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) type Provider struct { @@ -23,8 +24,8 @@ type Provider struct { } // NewProvider creates a new AWS provider instance. -func NewProvider(ctx context.Context, cfg types.ProviderConfig) (types.Provider, error) { - log := logging.New(&logging.Config{}).WithField("cloudprovider", "aws") +func NewProvider(ctx context.Context, log *logging.Logger, cfg types.ProviderConfig) (types.Provider, error) { + log = log.WithField("cloudprovider", "aws") awsConfig, err := buildAWSConfig(ctx, cfg) if err != nil { diff --git a/pkg/cloudprovider/aws/test/integration_test.go b/pkg/cloudprovider/aws/test/integration_test.go index 4688450b2..ee57d5f3b 100644 --- a/pkg/cloudprovider/aws/test/integration_test.go +++ b/pkg/cloudprovider/aws/test/integration_test.go @@ -54,7 +54,7 @@ func TestGetStorageState(t *testing.T) { cfg := getTestConfig(t) ctx := t.Context() - provider, err := aws.NewProvider(ctx, cfg) + provider, err := aws.NewProvider(ctx, nil, cfg) if err != nil { t.Fatalf("NewProvider failed: %v", err) } @@ -96,7 +96,7 @@ func TestRefreshNetworkState(t *testing.T) { cfg := getTestConfig(t) ctx := context.Background() - provider, err := aws.NewProvider(ctx, cfg) + provider, err := aws.NewProvider(ctx, nil, cfg) if err != nil { t.Fatalf("NewProvider failed: %v", err) } diff --git a/pkg/cloudprovider/gcp/provider.go b/pkg/cloudprovider/gcp/provider.go index 2d3040262..652d430c7 100644 --- a/pkg/cloudprovider/gcp/provider.go +++ b/pkg/cloudprovider/gcp/provider.go @@ -7,8 +7,9 @@ import ( "sync" compute "cloud.google.com/go/compute/apiv1" + "github.com/castai/kvisor/pkg/cloudprovider/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) type Provider struct { @@ -26,8 +27,8 @@ type Provider struct { } // NewProvider creates a new GCP provider instance. -func NewProvider(ctx context.Context, cfg types.ProviderConfig) (types.Provider, error) { - log := logging.New(&logging.Config{}).WithField("cloudprovider", "gcp") +func NewProvider(ctx context.Context, log *logging.Logger, cfg types.ProviderConfig) (types.Provider, error) { + log = log.WithField("cloudprovider", "gcp") // Build client options with authentication clientOptions, err := buildClientOptions(cfg) diff --git a/pkg/cloudprovider/gcp/test/integration_test.go b/pkg/cloudprovider/gcp/test/integration_test.go index 342fd1d42..df0075096 100644 --- a/pkg/cloudprovider/gcp/test/integration_test.go +++ b/pkg/cloudprovider/gcp/test/integration_test.go @@ -12,6 +12,7 @@ import ( "github.com/castai/kvisor/pkg/cloudprovider/gcp" "github.com/castai/kvisor/pkg/cloudprovider/types" + "github.com/castai/logging" ) // Use .env file or run tests with environment variables: @@ -48,7 +49,7 @@ func TestRefreshNetworkState(t *testing.T) { cfg := getTestConfig(t) ctx := context.Background() - provider, err := gcp.NewProvider(ctx, cfg) + provider, err := gcp.NewProvider(ctx, logging.New(), cfg) if err != nil { t.Fatalf("NewProvider failed: %v", err) } @@ -91,7 +92,7 @@ func TestGetStorageState(t *testing.T) { cfg := getTestConfig(t) ctx := t.Context() - provider, err := gcp.NewProvider(ctx, cfg) + provider, err := gcp.NewProvider(ctx, logging.New(), cfg) if err != nil { t.Fatalf("NewProvider failed: %v", err) } diff --git a/pkg/cloudprovider/provider.go b/pkg/cloudprovider/provider.go index 8d75f2e78..6057bb878 100644 --- a/pkg/cloudprovider/provider.go +++ b/pkg/cloudprovider/provider.go @@ -8,18 +8,19 @@ import ( "github.com/castai/kvisor/pkg/cloudprovider/aws" "github.com/castai/kvisor/pkg/cloudprovider/gcp" "github.com/castai/kvisor/pkg/cloudprovider/types" + "github.com/castai/logging" ) // TODO(samu): Remove when all providers are implemented var ErrProviderNotImplemented = errors.New("provider not implemented yet") // NewProvider creates a cloud provider instance based on config. -func NewProvider(ctx context.Context, cfg types.ProviderConfig) (types.Provider, error) { +func NewProvider(ctx context.Context, log *logging.Logger, cfg types.ProviderConfig) (types.Provider, error) { switch cfg.Type { case types.TypeGCP: - return gcp.NewProvider(ctx, cfg) + return gcp.NewProvider(ctx, log, cfg) case types.TypeAWS: - return aws.NewProvider(ctx, cfg) + return aws.NewProvider(ctx, log, cfg) case types.TypeAzure: return nil, fmt.Errorf("azure: %w", ErrProviderNotImplemented) default: diff --git a/pkg/containers/client.go b/pkg/containers/client.go index f8fc6fed7..47812a89f 100644 --- a/pkg/containers/client.go +++ b/pkg/containers/client.go @@ -24,8 +24,8 @@ import ( "github.com/castai/kvisor/cmd/agent/daemon/metrics" "github.com/castai/kvisor/pkg/cgroup" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" ) var ( diff --git a/pkg/containers/client_test.go b/pkg/containers/client_test.go index 1e0bbb074..0008aca4e 100644 --- a/pkg/containers/client_test.go +++ b/pkg/containers/client_test.go @@ -8,7 +8,7 @@ import ( "time" "github.com/castai/kvisor/pkg/cgroup" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/containerd/containerd" "github.com/containerd/containerd/api/services/tasks/v1" "github.com/containerd/containerd/content" @@ -253,7 +253,7 @@ func TestClient(t *testing.T) { func newTestClient() *Client { return &Client{ - log: logging.NewTestLog(), + log: logging.New(), criRuntimeServiceClient: &mockCriClient{}, cgroupClient: &mockCgroupClient{}, containersByID: map[string]*Container{}, diff --git a/pkg/ebpftracer/debug/debug.go b/pkg/ebpftracer/debug/debug.go index fa4bc2835..04f6cc597 100644 --- a/pkg/ebpftracer/debug/debug.go +++ b/pkg/ebpftracer/debug/debug.go @@ -12,8 +12,8 @@ import ( "sync/atomic" "github.com/castai/kvisor/pkg/ebpftracer/helpers" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" "github.com/cilium/ebpf" "github.com/cilium/ebpf/link" ) diff --git a/pkg/ebpftracer/decoder/decoder.go b/pkg/ebpftracer/decoder/decoder.go index 6c05ee745..e14e3e655 100644 --- a/pkg/ebpftracer/decoder/decoder.go +++ b/pkg/ebpftracer/decoder/decoder.go @@ -16,8 +16,8 @@ import ( castpb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/net/packet" + "github.com/castai/logging" "github.com/miekg/dns" "golang.org/x/sys/unix" ) diff --git a/pkg/ebpftracer/decoder/decoder_test.go b/pkg/ebpftracer/decoder/decoder_test.go index 35b046370..36b75abf4 100644 --- a/pkg/ebpftracer/decoder/decoder_test.go +++ b/pkg/ebpftracer/decoder/decoder_test.go @@ -9,7 +9,7 @@ import ( "testing" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/net/dns/dnsmessage" @@ -882,7 +882,7 @@ func TestDecodeSSH(t *testing.T) { }, } - log := logging.New(&logging.Config{}) + log := logging.New() for _, test := range testCases { t.Run(test.title, func(t *testing.T) { diff --git a/pkg/ebpftracer/metrics.go b/pkg/ebpftracer/metrics.go index e6be98252..3a4f00eb8 100644 --- a/pkg/ebpftracer/metrics.go +++ b/pkg/ebpftracer/metrics.go @@ -11,7 +11,7 @@ import ( "time" "github.com/castai/kvisor/cmd/agent/daemon/metrics" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/cilium/ebpf" ) diff --git a/pkg/ebpftracer/module.go b/pkg/ebpftracer/module.go index 56fe31577..1555d2d93 100644 --- a/pkg/ebpftracer/module.go +++ b/pkg/ebpftracer/module.go @@ -13,7 +13,7 @@ import ( "sync/atomic" "github.com/castai/kvisor/pkg/ebpftracer/helpers" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/cilium/ebpf" "github.com/cilium/ebpf/btf" "github.com/cilium/ebpf/rlimit" diff --git a/pkg/ebpftracer/policy_filters.go b/pkg/ebpftracer/policy_filters.go index 9332978ec..012d862a7 100644 --- a/pkg/ebpftracer/policy_filters.go +++ b/pkg/ebpftracer/policy_filters.go @@ -7,8 +7,8 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/decoder" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/net/iputil" + "github.com/castai/logging" "github.com/cespare/xxhash/v2" "github.com/elastic/go-freelru" "github.com/samber/lo" diff --git a/pkg/ebpftracer/policy_filters_test.go b/pkg/ebpftracer/policy_filters_test.go index 891abb858..05592235b 100644 --- a/pkg/ebpftracer/policy_filters_test.go +++ b/pkg/ebpftracer/policy_filters_test.go @@ -9,7 +9,7 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/decoder" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) @@ -70,7 +70,7 @@ func TestFilterAnd(t *testing.T) { func TestDNSPolicyFilter(t *testing.T) { r := require.New(t) - log := logging.NewTestLog() + log := logging.New() f := DeduplicateDNSEventsPreFilter(log, 2, 1*time.Hour) g := f() diff --git a/pkg/ebpftracer/signature/git_clone_detected.go b/pkg/ebpftracer/signature/git_clone_detected.go index 4ba4f9df3..55cc9bc2a 100644 --- a/pkg/ebpftracer/signature/git_clone_detected.go +++ b/pkg/ebpftracer/signature/git_clone_detected.go @@ -7,7 +7,7 @@ import ( v1 "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) var _ Signature = (*GitCloneDetected)(nil) diff --git a/pkg/ebpftracer/signature/git_clone_detected_test.go b/pkg/ebpftracer/signature/git_clone_detected_test.go index ebea1f75b..dfa882656 100644 --- a/pkg/ebpftracer/signature/git_clone_detected_test.go +++ b/pkg/ebpftracer/signature/git_clone_detected_test.go @@ -7,7 +7,7 @@ import ( "github.com/castai/kvisor/pkg/containers" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) @@ -308,7 +308,7 @@ func TestGitCloneDetectedSignature(t *testing.T) { }, } - log := logging.New(&logging.Config{}) + log := logging.New() for _, test := range testCases { t.Run(test.title, func(t *testing.T) { diff --git a/pkg/ebpftracer/signature/ingress_nightmare_detected.go b/pkg/ebpftracer/signature/ingress_nightmare_detected.go index 6fb02e287..8a4cc147a 100644 --- a/pkg/ebpftracer/signature/ingress_nightmare_detected.go +++ b/pkg/ebpftracer/signature/ingress_nightmare_detected.go @@ -6,7 +6,7 @@ import ( v1 "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "golang.org/x/sys/unix" ) diff --git a/pkg/ebpftracer/signature/signature.go b/pkg/ebpftracer/signature/signature.go index ae8581bfa..d79e01d2f 100644 --- a/pkg/ebpftracer/signature/signature.go +++ b/pkg/ebpftracer/signature/signature.go @@ -6,7 +6,7 @@ import ( castpb "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) type SignatureMetadata struct { diff --git a/pkg/ebpftracer/signature/signature_rules.go b/pkg/ebpftracer/signature/signature_rules.go index 48455ba42..41b9bf623 100644 --- a/pkg/ebpftracer/signature/signature_rules.go +++ b/pkg/ebpftracer/signature/signature_rules.go @@ -1,6 +1,6 @@ package signature -import "github.com/castai/kvisor/pkg/logging" +import "github.com/castai/logging" type DefaultSignatureConfig struct { SOCKS5DetectedSignatureEnabled bool `json:"SOCKS5DetectedSignatureEnabled"` diff --git a/pkg/ebpftracer/signature/socks5_detected.go b/pkg/ebpftracer/signature/socks5_detected.go index f3ab4dbba..b077b28cf 100644 --- a/pkg/ebpftracer/signature/socks5_detected.go +++ b/pkg/ebpftracer/signature/socks5_detected.go @@ -4,9 +4,9 @@ import ( v1 "github.com/castai/kvisor/api/v1/runtime" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/net/packet" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" "github.com/elastic/go-freelru" ) diff --git a/pkg/ebpftracer/signature/socks5_detected_test.go b/pkg/ebpftracer/signature/socks5_detected_test.go index eb4d39987..2d2213f9b 100644 --- a/pkg/ebpftracer/signature/socks5_detected_test.go +++ b/pkg/ebpftracer/signature/socks5_detected_test.go @@ -7,7 +7,7 @@ import ( "github.com/castai/kvisor/pkg/containers" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) @@ -341,7 +341,7 @@ func TestSOCKS5DetectedSignature(t *testing.T) { }, } - log := logging.New(&logging.Config{}) + log := logging.New() for _, test := range testCases { t.Run(test.title, func(t *testing.T) { diff --git a/pkg/ebpftracer/tracer.go b/pkg/ebpftracer/tracer.go index a7255e90e..5c6fa4f18 100644 --- a/pkg/ebpftracer/tracer.go +++ b/pkg/ebpftracer/tracer.go @@ -15,10 +15,10 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/signature" "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/castai/kvisor/pkg/system" + "github.com/castai/logging" "github.com/elastic/go-freelru" "github.com/go-playground/validator/v10" "github.com/google/gopacket/layers" diff --git a/pkg/ebpftracer/tracer_decode_benchmark_test.go b/pkg/ebpftracer/tracer_decode_benchmark_test.go index b4c3b1d3b..55b64f68c 100644 --- a/pkg/ebpftracer/tracer_decode_benchmark_test.go +++ b/pkg/ebpftracer/tracer_decode_benchmark_test.go @@ -6,7 +6,7 @@ import ( "testing" "github.com/castai/kvisor/pkg/ebpftracer/decoder" - "github.com/castai/kvisor/pkg/logging" + "github.com/castai/logging" ) func BenchmarkFilterDecodeAndExportEvent(b *testing.B) { @@ -21,7 +21,7 @@ func BenchmarkFilterDecodeAndExportEvent(b *testing.B) { if err != nil { b.Fatal(err) } - dec := decoder.NewEventDecoder(logging.NewTestLog(), []byte{}) + dec := decoder.NewEventDecoder(logging.New(), []byte{}) tracer := buildTestTracer() @@ -50,7 +50,7 @@ func BenchmarkModuleLoad(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { - mo := newModule(logging.NewTestLog()) + mo := newModule(logging.New()) if err := mo.load(Config{ BTFPath: "/sys/kernel/btf/vmlinux", SignalEventsRingBufferSize: 4096, diff --git a/pkg/ebpftracer/tracer_decode_test.go b/pkg/ebpftracer/tracer_decode_test.go index b62f23c94..b0c998e68 100644 --- a/pkg/ebpftracer/tracer_decode_test.go +++ b/pkg/ebpftracer/tracer_decode_test.go @@ -14,8 +14,8 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/decoder" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" "github.com/stretchr/testify/require" ) @@ -134,7 +134,7 @@ func TestFilterDecodeAndExportEvent(t *testing.T) { applyTestPolicy(tracer, tc.policy) } - dec := decoder.NewEventDecoder(logging.NewTestLog(), testEventData) + dec := decoder.NewEventDecoder(logging.New(), testEventData) err := tracer.decodeAndExportEvent(context.TODO(), dec) require.NoError(t, err) @@ -153,7 +153,7 @@ func TestDecodeMagicWriteEvent(t *testing.T) { data, err := os.ReadFile(path) r.NoError(err) - dec := decoder.NewEventDecoder(logging.New(&logging.Config{}), data) + dec := decoder.NewEventDecoder(logging.New(), data) tr := &Tracer{ eventsChan: make(chan *types.Event), @@ -199,7 +199,7 @@ func TestDecodeSchedProcessExecEvent(t *testing.T) { }.Encode() r.NoError(err) - dec := decoder.NewEventDecoder(logging.New(&logging.Config{}), eventCtx) + dec := decoder.NewEventDecoder(logging.New(), eventCtx) go func() { err = tr.decodeAndExportEvent(context.Background(), dec) @@ -243,7 +243,7 @@ func TestDecodeSchedProcessExitEvent(t *testing.T) { eventCtx, err := exitEvent.Encode() r.NoError(err) - dec := decoder.NewEventDecoder(logging.New(&logging.Config{}), eventCtx) + dec := decoder.NewEventDecoder(logging.New(), eventCtx) go func() { err = tr.decodeAndExportEvent(context.Background(), dec) @@ -261,7 +261,7 @@ func TestDecodeSchedProcessExitEvent(t *testing.T) { eventCtx, err := exitEvent.Encode() r.NoError(err) - dec := decoder.NewEventDecoder(logging.New(&logging.Config{}), eventCtx) + dec := decoder.NewEventDecoder(logging.New(), eventCtx) go func() { err = tr.decodeAndExportEvent(context.Background(), dec) diff --git a/pkg/ebpftracer/tracer_filter_test.go b/pkg/ebpftracer/tracer_filter_test.go index 1b0b070d9..4fdd09928 100644 --- a/pkg/ebpftracer/tracer_filter_test.go +++ b/pkg/ebpftracer/tracer_filter_test.go @@ -2,18 +2,18 @@ package ebpftracer import ( "context" - "log/slog" "testing" + "github.com/google/gopacket/layers" + "github.com/stretchr/testify/require" + "github.com/castai/kvisor/pkg/cgroup" "github.com/castai/kvisor/pkg/containers" "github.com/castai/kvisor/pkg/ebpftracer/decoder" "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/types" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/processtree" - "github.com/google/gopacket/layers" - "github.com/stretchr/testify/require" + "github.com/castai/logging" ) func TestAllowedByPolicyShouldBePerCgroup(t *testing.T) { @@ -151,9 +151,7 @@ func (c *MockContainerClient) CleanupByCgroupID(cgroup uint64) { type tracerOption func(*Tracer) func buildTestTracer(options ...tracerOption) *Tracer { - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) + log := logging.New() tracer := &Tracer{ log: log, diff --git a/pkg/ebpftracer/tracer_playground_test.go b/pkg/ebpftracer/tracer_playground_test.go index a2166d0d4..c632d384e 100644 --- a/pkg/ebpftracer/tracer_playground_test.go +++ b/pkg/ebpftracer/tracer_playground_test.go @@ -5,7 +5,6 @@ import ( "context" "errors" "fmt" - "log/slog" "net" "net/netip" "os" @@ -13,6 +12,12 @@ import ( "testing" "time" + "github.com/davecgh/go-spew/spew" + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + "golang.org/x/sync/errgroup" + "golang.org/x/sys/unix" + "github.com/castai/kvisor/pkg/cgroup" "github.com/castai/kvisor/pkg/containers" "github.com/castai/kvisor/pkg/ebpftracer" @@ -20,14 +25,9 @@ import ( "github.com/castai/kvisor/pkg/ebpftracer/events" "github.com/castai/kvisor/pkg/ebpftracer/signature" "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/davecgh/go-spew/spew" - "github.com/google/gopacket" - "github.com/google/gopacket/layers" - "golang.org/x/sync/errgroup" - "golang.org/x/sys/unix" + "github.com/castai/logging" ) func TestTracer(t *testing.T) { @@ -37,9 +37,7 @@ func TestTracer(t *testing.T) { ctx := context.Background() - log := logging.New(&logging.Config{ - Level: slog.LevelDebug, - }) + log := logging.New() procHandle := proc.New() pidNS, err := procHandle.GetCurrentPIDNSID() diff --git a/pkg/ebpftracer/tracer_test.go b/pkg/ebpftracer/tracer_test.go index 56b066c30..401801493 100644 --- a/pkg/ebpftracer/tracer_test.go +++ b/pkg/ebpftracer/tracer_test.go @@ -2,19 +2,22 @@ package ebpftracer import ( "bytes" + "log/slog" "testing" - "github.com/castai/kvisor/pkg/logging" "github.com/stretchr/testify/require" + + "github.com/castai/logging" ) func TestTracer(t *testing.T) { t.Run("log ebpf issue as warning", func(t *testing.T) { r := require.New(t) logOut := bytes.NewBuffer(nil) - log := logging.New(&logging.Config{ + log := logging.New(logging.NewTextHandler(logging.TextHandlerConfig{ + Level: slog.LevelDebug, Output: logOut, - }) + })) tr := &Tracer{ log: log, currentTracerEbpfMetrics: map[string]uint64{}, diff --git a/pkg/logging/export_handler.go b/pkg/logging/export_handler.go deleted file mode 100644 index 15fa21629..000000000 --- a/pkg/logging/export_handler.go +++ /dev/null @@ -1,65 +0,0 @@ -package logging - -import ( - "context" - "log/slog" -) - -type ExportFunc func(ctx context.Context, record slog.Record) - -func NewExportHandler(ctx context.Context, next slog.Handler, cfg ExportConfig) slog.Handler { - handler := &ExportHandler{ - next: next, - cfg: cfg, - ch: make(chan slog.Record, 1000), - } - go handler.run(ctx) - - return handler -} - -type ExportHandler struct { - next slog.Handler - cfg ExportConfig - - ch chan slog.Record -} - -func (e *ExportHandler) Enabled(ctx context.Context, level slog.Level) bool { - return e.next.Enabled(ctx, level) -} - -func (e *ExportHandler) Handle(ctx context.Context, record slog.Record) error { - if record.Level >= e.cfg.MinLevel { - e.cfg.ExportFunc(ctx, record) - } - - return e.next.Handle(ctx, record) -} - -func (e *ExportHandler) WithAttrs(attrs []slog.Attr) slog.Handler { - return &ExportHandler{ - next: e.next.WithAttrs(attrs), - cfg: e.cfg, - ch: e.ch, - } -} - -func (e *ExportHandler) WithGroup(name string) slog.Handler { - return &ExportHandler{ - next: e.next.WithGroup(name), - cfg: e.cfg, - ch: e.ch, - } -} - -func (e *ExportHandler) run(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case record := <-e.ch: - e.cfg.ExportFunc(ctx, record) - } - } -} diff --git a/pkg/logging/logging.go b/pkg/logging/logging.go deleted file mode 100644 index 57aceef7f..000000000 --- a/pkg/logging/logging.go +++ /dev/null @@ -1,159 +0,0 @@ -package logging - -import ( - "context" - "fmt" - "io" - "log/slog" - "os" - "path/filepath" - "runtime" - "strings" - "time" - - "golang.org/x/time/rate" -) - -type Config struct { - Ctx context.Context - RateLimiter RateLimiterConfig - Level slog.Level - AddSource bool - Output io.Writer - Export ExportConfig -} - -type RateLimiterConfig struct { - Limit rate.Limit - Burst int - Inform bool -} - -type ExportConfig struct { - ExportFunc ExportFunc - MinLevel slog.Level -} - -func MustParseLevel(lvlStr string) slog.Level { - var lvl slog.Level - err := lvl.UnmarshalText([]byte(lvlStr)) - if err != nil { - panic("parsing log level from level string " + lvlStr) - } - return lvl -} - -func init() { - slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{}))) -} - -func New(cfg *Config) *Logger { - out := cfg.Output - if out == nil { - out = os.Stdout - } - if cfg.Ctx == nil { - cfg.Ctx = context.Background() - } - - // Initial logger. - var handler slog.Handler = slog.NewTextHandler(out, &slog.HandlerOptions{ - AddSource: cfg.AddSource, - Level: cfg.Level, - ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr { - if a.Key == slog.LevelKey { - a.Value = slog.StringValue(strings.ToLower(a.Value.String())) - } - if cfg.AddSource { - // Remove the directory from the source's filename. - if a.Key == slog.SourceKey { - source := a.Value.Any().(*slog.Source) - source.File = filepath.Base(source.File) - } - } - return a - }, - }) - - // Export logs handler. - if cfg.Export.ExportFunc != nil { - handler = NewExportHandler(cfg.Ctx, handler, cfg.Export) - } - - // Rate limiter handler. - if cfg.RateLimiter.Limit != 0 { - handler = NewRateLimiterHandler(cfg.Ctx, handler, cfg.RateLimiter) - } - - log := slog.New(handler) - return &Logger{log: log} -} - -func NewTestLog() *Logger { - return New(&Config{Level: slog.LevelDebug}) -} - -type Logger struct { - log *slog.Logger -} - -func (l *Logger) Error(msg string) { - l.doLog(slog.LevelError, msg) //nolint:govet -} - -func (l *Logger) Errorf(format string, a ...any) { - l.doLog(slog.LevelError, fmt.Sprintf(format, a...)) -} - -func (l *Logger) Infof(format string, a ...any) { - l.doLog(slog.LevelInfo, fmt.Sprintf(format, a...)) -} - -func (l *Logger) Info(msg string) { - l.doLog(slog.LevelInfo, msg) //nolint:govet -} - -func (l *Logger) Debug(msg string) { - l.doLog(slog.LevelDebug, msg) //nolint:govet -} - -func (l *Logger) Debugf(format string, a ...any) { - l.doLog(slog.LevelDebug, fmt.Sprintf(format, a...)) -} - -func (l *Logger) Warn(msg string) { - l.doLog(slog.LevelWarn, msg) //nolint:govet -} - -func (l *Logger) Warnf(format string, a ...any) { - l.doLog(slog.LevelWarn, fmt.Sprintf(format, a...)) -} - -func (l *Logger) Fatal(msg string) { - l.doLog(slog.LevelError, msg) //nolint:govet - os.Exit(1) -} - -func (l *Logger) IsEnabled(lvl slog.Level) bool { - ctx := context.Background() - return l.log.Handler().Enabled(ctx, lvl) -} - -func (l *Logger) doLog(lvl slog.Level, msg string) { - ctx := context.Background() - if !l.log.Handler().Enabled(ctx, lvl) { - return - } - var pcs [1]uintptr - runtime.Callers(3, pcs[:]) - r := slog.NewRecord(time.Now(), lvl, msg, pcs[0]) - _ = l.log.Handler().Handle(ctx, r) //nolint:contextcheck -} - -func (l *Logger) With(args ...any) *Logger { - return &Logger{log: l.log.With(args...)} -} - -func (l *Logger) WithField(k, v string) *Logger { - return &Logger{log: l.log.With(slog.String(k, v))} -} diff --git a/pkg/logging/logging_test.go b/pkg/logging/logging_test.go deleted file mode 100644 index ea4da9272..000000000 --- a/pkg/logging/logging_test.go +++ /dev/null @@ -1,80 +0,0 @@ -package logging_test - -import ( - "bytes" - "context" - "errors" - "log/slog" - "testing" - "time" - - "github.com/castai/kvisor/pkg/logging" - "github.com/stretchr/testify/require" - "golang.org/x/time/rate" -) - -func TestLogger(t *testing.T) { - t.Run("print long", func(t *testing.T) { - log := logging.New(&logging.Config{ - Level: logging.MustParseLevel("DEBUG"), - AddSource: true, - }) - - log.Errorf("something wrong: %v", errors.New("ups")) - serverLog := log.WithField("component", "server") - serverLog.Info("with component") - serverLog.Info("more server logs") - }) - - t.Run("rate limit", func(t *testing.T) { - var out bytes.Buffer - log := logging.New(&logging.Config{ - Output: &out, - Level: logging.MustParseLevel("DEBUG"), - RateLimiter: logging.RateLimiterConfig{ - Limit: rate.Every(10 * time.Millisecond), - Burst: 1, - }, - }) - - for i := 0; i < 10; i++ { - log.WithField("component", "test").Info("test") - time.Sleep(8 * time.Millisecond) - } - - require.GreaterOrEqual(t, countLogLines(&out), 5) - }) - - t.Run("export logs", func(t *testing.T) { - exportedLogs := make(chan slog.Record, 1) - log := logging.New(&logging.Config{ - Level: logging.MustParseLevel("DEBUG"), - Export: logging.ExportConfig{ - ExportFunc: func(ctx context.Context, record slog.Record) { - exportedLogs <- record - }, - MinLevel: slog.LevelInfo, - }, - }) - - //log.Debug("should not export debug") - log.WithField("component", "test").Error("should export error") - - select { - case logRecord := <-exportedLogs: - require.Equal(t, logRecord.Message, "should export error") - case <-time.After(time.Second): - t.Fatal("timeout") - } - }) -} - -func countLogLines(buf *bytes.Buffer) int { - var n int - for _, b := range buf.Bytes() { - if b == '\n' { - n++ - } - } - return n -} diff --git a/pkg/logging/ratelimiter_handler.go b/pkg/logging/ratelimiter_handler.go deleted file mode 100644 index ca585dfde..000000000 --- a/pkg/logging/ratelimiter_handler.go +++ /dev/null @@ -1,91 +0,0 @@ -package logging - -import ( - "context" - "fmt" - "log/slog" - "sync/atomic" - "time" - - "golang.org/x/time/rate" -) - -func NewRateLimiterHandler(ctx context.Context, next slog.Handler, cfg RateLimiterConfig) slog.Handler { - droppedLogsCounters := map[slog.Level]*atomic.Uint64{ - slog.LevelDebug: {}, - slog.LevelInfo: {}, - slog.LevelWarn: {}, - slog.LevelError: {}, - } - logsRate := cfg.Limit - burst := cfg.Burst - if cfg.Inform { - go printDroppedLogsCounter(ctx, droppedLogsCounters) - } - return &RateLimiterHandler{ - next: next, - rt: map[slog.Level]*rate.Limiter{ - slog.LevelDebug: rate.NewLimiter(logsRate, burst), - slog.LevelInfo: rate.NewLimiter(logsRate, burst), - slog.LevelWarn: rate.NewLimiter(logsRate, burst), - slog.LevelError: rate.NewLimiter(logsRate, burst), - }, - droppedLogsCounters: droppedLogsCounters, - } -} - -type RateLimiterHandler struct { - next slog.Handler - rt map[slog.Level]*rate.Limiter - droppedLogsCounters map[slog.Level]*atomic.Uint64 -} - -func (s *RateLimiterHandler) Enabled(ctx context.Context, level slog.Level) bool { - if !s.next.Enabled(ctx, level) { - return false - } - if !s.rt[level].Allow() { - s.droppedLogsCounters[level].Add(1) - return false - } - return true -} - -func (s *RateLimiterHandler) Handle(ctx context.Context, record slog.Record) error { - return s.next.Handle(ctx, record) -} - -func (s *RateLimiterHandler) WithAttrs(attrs []slog.Attr) slog.Handler { - return &RateLimiterHandler{ - next: s.next.WithAttrs(attrs), - rt: s.rt, - droppedLogsCounters: s.droppedLogsCounters, - } -} - -func (s *RateLimiterHandler) WithGroup(name string) slog.Handler { - return &RateLimiterHandler{ - next: s.next.WithGroup(name), - rt: s.rt, - droppedLogsCounters: s.droppedLogsCounters, - } -} - -func printDroppedLogsCounter(ctx context.Context, droppedLogsCounters map[slog.Level]*atomic.Uint64) { - ticker := time.NewTicker(5 * time.Second) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - for level, val := range droppedLogsCounters { - count := val.Load() - if count > 0 { - slog.Warn(fmt.Sprintf("logs rate limit, dropped %d lines for level %s", count, level.String())) - val.Store(0) - } - } - } - } -} diff --git a/pkg/processtree/processtree.go b/pkg/processtree/processtree.go index 459633fe2..d28734f0a 100644 --- a/pkg/processtree/processtree.go +++ b/pkg/processtree/processtree.go @@ -8,8 +8,8 @@ import ( "time" "github.com/castai/kvisor/pkg/containers" - "github.com/castai/kvisor/pkg/logging" "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" ) type ProcessAction int diff --git a/pkg/processtree/processtree_test.go b/pkg/processtree/processtree_test.go index cb1946444..140a705b5 100644 --- a/pkg/processtree/processtree_test.go +++ b/pkg/processtree/processtree_test.go @@ -6,11 +6,12 @@ import ( "testing" "time" - "github.com/castai/kvisor/pkg/containers" - "github.com/castai/kvisor/pkg/logging" - "github.com/castai/kvisor/pkg/proc" "github.com/samber/lo" "github.com/stretchr/testify/require" + + "github.com/castai/kvisor/pkg/containers" + "github.com/castai/kvisor/pkg/proc" + "github.com/castai/logging" ) func TestInitProcessTree(t *testing.T) { @@ -22,7 +23,7 @@ func TestInitProcessTree(t *testing.T) { containerID1 := "container-1" containerID2 := "container-2" - tree := New(logging.New(&logging.Config{}), proc.NewFromFS(fs.(proc.ProcFS)), + tree := New(logging.New(), proc.NewFromFS(fs.(proc.ProcFS)), &dummyContainerClient{ loadContainerTaskFn: func(ctx context.Context) ([]containers.ContainerProcess, error) { return []containers.ContainerProcess{