From 2d1e729473bd6b4c4bcb0e961bd08f1b7d65e1dc Mon Sep 17 00:00:00 2001 From: jt-dd <112463504+jt-dd@users.noreply.github.com> Date: Thu, 14 Nov 2024 18:21:29 +0100 Subject: [PATCH] Fix observability (#285) * Fix observability * fix split branches * fix linter * fix split branches * fix linter * Update pkg/telemetry/events/events.go Co-authored-by: Edouard Schweisguth * rework event api * PR comment * unifying events API --------- Co-authored-by: Edouard Schweisguth --- datadog/span-metrics.tf | 10 +-- pkg/collector/file.go | 14 +-- pkg/collector/k8s_api.go | 22 ++--- pkg/ingestor/api/api.go | 33 +++---- pkg/ingestor/puller/blob/blob.go | 48 ++++++---- pkg/kubehound/core/core_dump.go | 18 +--- pkg/kubehound/graph/builder.go | 2 +- pkg/kubehound/providers/providers.go | 18 ++++ .../storage/cache/memcache_provider.go | 4 +- .../storage/cache/memcache_writer.go | 4 +- .../storage/graphdb/janusgraph_edge_writer.go | 10 +-- .../graphdb/janusgraph_vertex_writer.go | 10 +-- pkg/kubehound/storage/storedb/mongo_writer.go | 8 +- pkg/telemetry/events/events.go | 90 +++++++++++++++++-- pkg/telemetry/log/fields.go | 12 +++ pkg/telemetry/metric/metrics.go | 7 ++ pkg/telemetry/span/spans.go | 16 ++-- pkg/telemetry/statsd/statsd.go | 30 ++++--- pkg/telemetry/tag/tags.go | 48 +++++++--- 19 files changed, 275 insertions(+), 129 deletions(-) diff --git a/datadog/span-metrics.tf b/datadog/span-metrics.tf index e5aea673c..68e8d6125 100644 --- a/datadog/span-metrics.tf +++ b/datadog/span-metrics.tf @@ -31,7 +31,7 @@ resource "datadog_spans_metric" "kubehound_ingest_duration" { } filter { - query = "service:kubehound operation_name:kubehound.ingestData" + query = "service:kubehound-ingestor operation_name:kubehound.ingestData" } dynamic "group_by" { @@ -63,7 +63,7 @@ resource "datadog_spans_metric" "kubehound_graph_duration" { } filter { - query = "service:kubehound operation_name:kubehound.buildGraph" + query = "service:kubehound-ingestor operation_name:kubehound.buildGraph" } dynamic "group_by" { @@ -103,7 +103,7 @@ resource "datadog_spans_metric" "kubehound_collector_stream_duration" { } filter { - query = "service:kubehound operation_name:kubehound.collector.stream" + query = "service:kubehound-collector operation_name:kubehound.collector.stream" } dynamic "group_by" { @@ -144,7 +144,7 @@ resource "datadog_spans_metric" "kubehound_graph_builder_edge_duration" { } filter { - query = "service:kubehound operation_name:kubehound.graph.builder.edge" + query = "service:kubehound-ingestor operation_name:kubehound.graph.builder.edge" } dynamic "group_by" { @@ -154,4 +154,4 @@ resource "datadog_spans_metric" "kubehound_graph_builder_edge_duration" { path = group_by.value } } -} \ No newline at end of file +} diff --git a/pkg/collector/file.go b/pkg/collector/file.go index 09172c60e..089431f58 100644 --- a/pkg/collector/file.go +++ b/pkg/collector/file.go @@ -124,7 +124,7 @@ func (c *FileCollector) streamPodsNamespace(ctx context.Context, fp string, inge } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.pod, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.pod, 1) i := item err = ingestor.IngestPod(ctx, &i) if err != nil { @@ -176,7 +176,7 @@ func (c *FileCollector) streamRolesNamespace(ctx context.Context, fp string, ing } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.role, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.role, 1) i := item err = ingestor.IngestRole(ctx, &i) if err != nil { @@ -228,7 +228,7 @@ func (c *FileCollector) streamRoleBindingsNamespace(ctx context.Context, fp stri } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.rolebinding, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.rolebinding, 1) i := item err = ingestor.IngestRoleBinding(ctx, &i) if err != nil { @@ -280,7 +280,7 @@ func (c *FileCollector) streamEndpointsNamespace(ctx context.Context, fp string, } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.endpoint, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.endpoint, 1) i := item err = ingestor.IngestEndpoint(ctx, &i) if err != nil { @@ -339,7 +339,7 @@ func (c *FileCollector) StreamNodes(ctx context.Context, ingestor NodeIngestor) } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.node, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.node, 1) i := item err = ingestor.IngestNode(ctx, &i) if err != nil { @@ -366,7 +366,7 @@ func (c *FileCollector) StreamClusterRoles(ctx context.Context, ingestor Cluster } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.clusterrole, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.clusterrole, 1) i := item err = ingestor.IngestClusterRole(ctx, &i) if err != nil { @@ -393,7 +393,7 @@ func (c *FileCollector) StreamClusterRoleBindings(ctx context.Context, ingestor } for _, item := range list.Items { - _ = statsd.Incr(metric.CollectorCount, c.tags.clusterrolebinding, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.clusterrolebinding, 1) i := item err = ingestor.IngestClusterRoleBinding(ctx, &i) if err != nil { diff --git a/pkg/collector/k8s_api.go b/pkg/collector/k8s_api.go index f6cc31996..ac88dbc6c 100644 --- a/pkg/collector/k8s_api.go +++ b/pkg/collector/k8s_api.go @@ -162,7 +162,7 @@ func (c *k8sAPICollector) wait(ctx context.Context, resourceType string, tags [] } // entity := tag.Entity(resourceType) - err := statsd.Gauge(metric.CollectorWait, float64(c.waitTime[resourceType]), tags, 1) + err := statsd.Gauge(ctx, metric.CollectorWait, float64(c.waitTime[resourceType]), tags, 1) if err != nil { l.Error("could not send gauge", log.ErrorField(err)) } @@ -213,19 +213,19 @@ func (c *k8sAPICollector) computeMetrics(ctx context.Context) (Metrics, error) { } runDuration := time.Since(c.startTime) - err := statsd.Gauge(metric.CollectorRunWait, float64(runTotalWaitTime), c.tags.baseTags, 1) + err := statsd.Gauge(ctx, metric.CollectorRunWait, float64(runTotalWaitTime), c.tags.baseTags, 1) if err != nil { errMetric = errors.Join(errMetric, err) l.Error("could not send gauge", log.ErrorField(err)) } - err = statsd.Gauge(metric.CollectorRunDuration, float64(runDuration), c.tags.baseTags, 1) + err = statsd.Gauge(ctx, metric.CollectorRunDuration, float64(runDuration), c.tags.baseTags, 1) if err != nil { errMetric = errors.Join(errMetric, err) l.Error("could not send gauge", log.ErrorField(err)) } runThrottlingPercentage := 1 - (float64(runDuration-runTotalWaitTime) / float64(runDuration)) - err = statsd.Gauge(metric.CollectorRunThrottling, runThrottlingPercentage, c.tags.baseTags, 1) + err = statsd.Gauge(ctx, metric.CollectorRunThrottling, runThrottlingPercentage, c.tags.baseTags, 1) if err != nil { errMetric = errors.Join(errMetric, err) l.Error("could not send gauge", log.ErrorField(err)) @@ -288,7 +288,7 @@ func (c *k8sAPICollector) streamPodsNamespace(ctx context.Context, namespace str c.setPagerConfig(pager) return pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.pod, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.pod, 1) c.wait(ctx, entity, c.tags.pod) item, ok := obj.(*corev1.Pod) if !ok { @@ -343,7 +343,7 @@ func (c *k8sAPICollector) streamRolesNamespace(ctx context.Context, namespace st c.setPagerConfig(pager) return pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.role, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.role, 1) c.wait(ctx, entity, c.tags.role) item, ok := obj.(*rbacv1.Role) if !ok { @@ -398,7 +398,7 @@ func (c *k8sAPICollector) streamRoleBindingsNamespace(ctx context.Context, names c.setPagerConfig(pager) return pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.rolebinding, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.rolebinding, 1) c.wait(ctx, entity, c.tags.rolebinding) item, ok := obj.(*rbacv1.RoleBinding) if !ok { @@ -453,7 +453,7 @@ func (c *k8sAPICollector) streamEndpointsNamespace(ctx context.Context, namespac c.setPagerConfig(pager) return pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.endpoint, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.endpoint, 1) c.wait(ctx, entity, c.tags.endpoint) item, ok := obj.(*discoveryv1.EndpointSlice) if !ok { @@ -507,7 +507,7 @@ func (c *k8sAPICollector) StreamNodes(ctx context.Context, ingestor NodeIngestor c.setPagerConfig(pager) err = pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.node, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.node, 1) c.wait(ctx, entity, c.tags.node) item, ok := obj.(*corev1.Node) if !ok { @@ -550,7 +550,7 @@ func (c *k8sAPICollector) StreamClusterRoles(ctx context.Context, ingestor Clust c.setPagerConfig(pager) err = pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.clusterrole, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.clusterrole, 1) c.wait(ctx, entity, c.tags.clusterrole) item, ok := obj.(*rbacv1.ClusterRole) if !ok { @@ -593,7 +593,7 @@ func (c *k8sAPICollector) StreamClusterRoleBindings(ctx context.Context, ingesto c.setPagerConfig(pager) err = pager.EachListItem(ctx, opts, func(obj runtime.Object) error { - _ = statsd.Incr(metric.CollectorCount, c.tags.clusterrolebinding, 1) + _ = statsd.Incr(ctx, metric.CollectorCount, c.tags.clusterrolebinding, 1) c.wait(ctx, entity, c.tags.clusterrolebinding) item, ok := obj.(*rbacv1.ClusterRoleBinding) if !ok { diff --git a/pkg/ingestor/api/api.go b/pkg/ingestor/api/api.go index aa8c248a0..3304e98ef 100644 --- a/pkg/ingestor/api/api.go +++ b/pkg/ingestor/api/api.go @@ -11,21 +11,19 @@ import ( "github.com/DataDog/KubeHound/pkg/collector" "github.com/DataDog/KubeHound/pkg/config" "github.com/DataDog/KubeHound/pkg/dump" - "github.com/DataDog/KubeHound/pkg/ingestor" grpc "github.com/DataDog/KubeHound/pkg/ingestor/api/grpc/pb" "github.com/DataDog/KubeHound/pkg/ingestor/notifier" "github.com/DataDog/KubeHound/pkg/ingestor/puller" - "github.com/DataDog/KubeHound/pkg/kubehound/graph" "github.com/DataDog/KubeHound/pkg/kubehound/graph/adapter" "github.com/DataDog/KubeHound/pkg/kubehound/providers" "github.com/DataDog/KubeHound/pkg/kubehound/store/collections" "github.com/DataDog/KubeHound/pkg/telemetry/events" "github.com/DataDog/KubeHound/pkg/telemetry/log" "github.com/DataDog/KubeHound/pkg/telemetry/span" - "github.com/DataDog/KubeHound/pkg/telemetry/tag" gremlingo "github.com/apache/tinkerpop/gremlin-go/v3/driver" "go.mongodb.org/mongo-driver/bson" "google.golang.org/protobuf/types/known/timestamppb" + "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext" "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer" ) @@ -153,27 +151,24 @@ func (g *IngestorAPI) Ingest(ctx context.Context, path string) error { runCtx := context.Background() runCtx = context.WithValue(runCtx, log.ContextFieldCluster, clusterName) runCtx = context.WithValue(runCtx, log.ContextFieldRunID, runID) - l = log.Logger(runCtx) //nolint: contextcheck - spanJob, runCtx := span.SpanRunFromContext(runCtx, span.IngestorStartJob) - defer func() { spanJob.Finish(tracer.WithError(err)) }() - - events.PushEvent( - fmt.Sprintf("Ingesting cluster %s with runID %s", clusterName, runID), - fmt.Sprintf("Ingesting cluster %s with runID %s", clusterName, runID), - []string{ - tag.IngestionRunID(runID), - }, - ) - + l = log.Logger(runCtx) //nolint: contextcheck alreadyIngested, err := g.isAlreadyIngestedInGraph(runCtx, clusterName, runID) //nolint: contextcheck if err != nil { return err } if alreadyIngested { + _ = events.PushEvent(runCtx, events.IngestSkip, "") //nolint: contextcheck + return fmt.Errorf("%w [%s:%s]", ErrAlreadyIngested, clusterName, runID) } + spanJob, runCtx := span.SpanRunFromContext(runCtx, span.IngestorStartJob) + spanJob.SetTag(ext.ManualKeep, true) + defer func() { spanJob.Finish(tracer.WithError(err)) }() + + _ = events.PushEvent(runCtx, events.IngestStarted, "") //nolint: contextcheck + // We need to flush the cache to prevent warnings/errors when overwriting elements in cache from the previous ingestion // This avoid conflicts from previous ingestion (there is no need to reuse the cache from a previous ingestion) l.Info("Preparing cache provider") @@ -209,15 +204,11 @@ func (g *IngestorAPI) Ingest(ctx context.Context, path string) error { } } - err = ingestor.IngestData(runCtx, runCfg, collect, g.providers.CacheProvider, g.providers.StoreProvider, g.providers.GraphProvider) //nolint: contextcheck - if err != nil { - return fmt.Errorf("raw data ingest: %w", err) - } - - err = graph.BuildGraph(runCtx, runCfg, g.providers.StoreProvider, g.providers.GraphProvider, g.providers.CacheProvider) //nolint: contextcheck + err = g.providers.IngestBuildData(runCtx, runCfg) //nolint: contextcheck if err != nil { return err } + err = g.notifier.Notify(runCtx, clusterName, runID) //nolint: contextcheck if err != nil { return fmt.Errorf("notifying: %w", err) diff --git a/pkg/ingestor/puller/blob/blob.go b/pkg/ingestor/puller/blob/blob.go index 0afae671f..808602a5d 100644 --- a/pkg/ingestor/puller/blob/blob.go +++ b/pkg/ingestor/puller/blob/blob.go @@ -23,6 +23,8 @@ import ( _ "gocloud.dev/blob/gcsblob" _ "gocloud.dev/blob/memblob" "gocloud.dev/blob/s3blob" + "gopkg.in/DataDog/dd-trace-go.v1/ddtrace" + "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer" ) @@ -51,6 +53,9 @@ func NewBlobStorage(cfg *config.KubehoundConfig, blobConfig *config.BlobConfig) } func (bs *BlobStore) openBucket(ctx context.Context) (*blob.Bucket, error) { + l := log.Logger(ctx) + l.Info("Opening bucket", log.String("bucket_name", bs.bucketName)) + urlStruct, err := url.Parse(bs.bucketName) if err != nil { return nil, err @@ -132,10 +137,15 @@ func (bs *BlobStore) ListFiles(ctx context.Context, prefix string, recursive boo // Pull pulls the data from the blob store (e.g: s3) and returns the path of the folder containing the archive func (bs *BlobStore) Put(outer context.Context, archivePath string, clusterName string, runID string) error { l := log.Logger(outer) - l.Info("Putting data on blob store bucket", log.String("bucket_name", bs.bucketName), log.String(log.FieldClusterKey, clusterName), log.String(log.FieldRunIDKey, runID)) - spanPut, ctx := span.SpanRunFromContext(outer, span.IngestorBlobPull) var err error - defer func() { spanPut.Finish(tracer.WithError(err)) }() + + // Triggering a span only when it is an actual run and not the rehydration process (download the kubehound dump to get the metadata) + if log.GetRunIDFromContext(outer) != "" { + var spanPut ddtrace.Span + spanPut, outer = span.SpanRunFromContext(outer, span.IngestorBlobPull) + defer func() { spanPut.Finish(tracer.WithError(err)) }() + } + l.Info("Putting data on blob store bucket", log.String("bucket_name", bs.bucketName), log.String(log.FieldClusterKey, clusterName), log.String(log.FieldRunIDKey, runID)) dumpResult, err := dump.NewDumpResult(clusterName, runID, true) if err != nil { @@ -143,7 +153,7 @@ func (bs *BlobStore) Put(outer context.Context, archivePath string, clusterName } key := dumpResult.GetFullPath() l.Info("Opening bucket", log.String("bucket_name", bs.bucketName)) - b, err := bs.openBucket(ctx) + b, err := bs.openBucket(outer) if err != nil { return err } @@ -157,7 +167,7 @@ func (bs *BlobStore) Put(outer context.Context, archivePath string, clusterName l.Info("Uploading archive from blob store", log.String("key", key)) w := bufio.NewReader(f) - err = b.Upload(ctx, key, w, &blob.WriterOptions{ + err = b.Upload(outer, key, w, &blob.WriterOptions{ ContentType: "application/gzip", }) if err != nil { @@ -175,13 +185,15 @@ func (bs *BlobStore) Put(outer context.Context, archivePath string, clusterName // Pull pulls the data from the blob store (e.g: s3) and returns the path of the folder containing the archive func (bs *BlobStore) Pull(outer context.Context, key string) (string, error) { l := log.Logger(outer) - l.Info("Pulling data from blob store bucket", log.String("bucket_name", bs.bucketName), log.String("key", key)) - spanPull, ctx := span.SpanRunFromContext(outer, span.IngestorBlobPull) var err error - defer func() { spanPull.Finish(tracer.WithError(err)) }() + if log.GetRunIDFromContext(outer) != "" { + var spanPull ddtrace.Span + spanPull, outer = span.SpanRunFromContext(outer, span.IngestorBlobPull) + defer func() { spanPull.Finish(tracer.WithError(err)) }() + } + l.Info("Pulling data from blob store bucket", log.String("bucket_name", bs.bucketName), log.String("key", key)) - l.Info("Opening bucket", log.String("bucket_name", bs.bucketName)) - b, err := bs.openBucket(ctx) + b, err := bs.openBucket(outer) if err != nil { return "", err } @@ -207,7 +219,7 @@ func (bs *BlobStore) Pull(outer context.Context, key string) (string, error) { l.Info("Downloading archive from blob store", log.String("key", key)) w := bufio.NewWriter(f) - err = b.Download(ctx, key, w, nil) + err = b.Download(outer, key, w, nil) if err != nil { return archivePath, err } @@ -221,9 +233,12 @@ func (bs *BlobStore) Pull(outer context.Context, key string) (string, error) { } func (bs *BlobStore) Extract(ctx context.Context, archivePath string) error { - spanExtract, _ := span.SpanRunFromContext(ctx, span.IngestorBlobExtract) var err error - defer func() { spanExtract.Finish(tracer.WithError(err)) }() + if log.GetRunIDFromContext(ctx) != "" { + var spanPull ddtrace.Span + spanPull, ctx = span.SpanRunFromContext(ctx, span.IngestorBlobExtract) + defer func() { spanPull.Finish(tracer.WithError(err)) }() + } basePath := filepath.Dir(archivePath) err = puller.CheckSanePath(archivePath, basePath) @@ -243,9 +258,12 @@ func (bs *BlobStore) Extract(ctx context.Context, archivePath string) error { // Once downloaded and processed, we should cleanup the disk so we can reduce the disk usage // required for large infrastructure func (bs *BlobStore) Close(ctx context.Context, archivePath string) error { - spanClose, _ := span.SpanRunFromContext(ctx, span.IngestorBlobClose) var err error - defer func() { spanClose.Finish(tracer.WithError(err)) }() + if log.GetRunIDFromContext(ctx) != "" { + var spanClose ddtrace.Span + spanClose, _ = span.SpanRunFromContext(ctx, span.IngestorBlobClose) + defer func() { spanClose.Finish(tracer.WithError(err)) }() + } path := filepath.Dir(archivePath) err = puller.CheckSanePath(archivePath, bs.cfg.Ingestor.TempDir) diff --git a/pkg/kubehound/core/core_dump.go b/pkg/kubehound/core/core_dump.go index 4c638b221..c6e1e588b 100644 --- a/pkg/kubehound/core/core_dump.go +++ b/pkg/kubehound/core/core_dump.go @@ -13,7 +13,6 @@ import ( "github.com/DataDog/KubeHound/pkg/telemetry/events" "github.com/DataDog/KubeHound/pkg/telemetry/log" "github.com/DataDog/KubeHound/pkg/telemetry/span" - "github.com/DataDog/KubeHound/pkg/telemetry/tag" "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext" "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer" ) @@ -43,13 +42,7 @@ func DumpCore(ctx context.Context, khCfg *config.KubehoundConfig, upload bool) ( khCfg.Collector.Type = config.CollectorTypeK8sAPI - events.PushEvent( - fmt.Sprintf("Starting KubeHound dump for %s", clusterName), - fmt.Sprintf("Starting KubeHound dump for %s", clusterName), - []string{ - tag.ActionType(events.DumperRun), - }, - ) + _ = events.PushEvent(ctx, events.DumpStarted, "") filePath, err := runLocalDump(ctx, khCfg) if err != nil { @@ -76,13 +69,8 @@ func DumpCore(ctx context.Context, khCfg *config.KubehoundConfig, upload bool) ( } } - events.PushEvent( - fmt.Sprintf("Finish KubeHound dump for %s", clusterName), - fmt.Sprintf("KubeHound dump run has been completed in %s", time.Since(start)), - []string{ - tag.ActionType(events.DumperRun), - }, - ) + text := fmt.Sprintf("KubeHound dump run has been completed in %s", time.Since(start)) + _ = events.PushEvent(ctx, events.DumpFinished, text) l.Info("KubeHound dump run has been completed", log.Duration("duration", time.Since(start))) return filePath, nil diff --git a/pkg/kubehound/graph/builder.go b/pkg/kubehound/graph/builder.go index 801316782..2cb89827c 100644 --- a/pkg/kubehound/graph/builder.go +++ b/pkg/kubehound/graph/builder.go @@ -54,7 +54,7 @@ func (b *Builder) HealthCheck(ctx context.Context) error { // buildEdge inserts a class of edges into the graph database. func (b *Builder) buildEdge(ctx context.Context, label string, e edge.Builder, oic *converter.ObjectIDConverter) error { - span, ctx := tracer.StartSpanFromContext(ctx, span.BuildEdge, tracer.Measured(), tracer.ResourceName(e.Label())) + span, ctx := span.StartSpanFromContext(ctx, span.BuildEdge, tracer.Measured(), tracer.ResourceName(e.Label())) span.SetTag(tag.LabelTag, e.Label()) var err error defer func() { span.Finish(tracer.WithError(err)) }() diff --git a/pkg/kubehound/providers/providers.go b/pkg/kubehound/providers/providers.go index 5d476d4ae..6b63b5da3 100644 --- a/pkg/kubehound/providers/providers.go +++ b/pkg/kubehound/providers/providers.go @@ -3,6 +3,7 @@ package providers import ( "context" "fmt" + "time" "github.com/DataDog/KubeHound/pkg/collector" "github.com/DataDog/KubeHound/pkg/config" @@ -11,7 +12,11 @@ import ( "github.com/DataDog/KubeHound/pkg/kubehound/storage/cache" "github.com/DataDog/KubeHound/pkg/kubehound/storage/graphdb" "github.com/DataDog/KubeHound/pkg/kubehound/storage/storedb" + "github.com/DataDog/KubeHound/pkg/telemetry/events" "github.com/DataDog/KubeHound/pkg/telemetry/log" + "github.com/DataDog/KubeHound/pkg/telemetry/metric" + "github.com/DataDog/KubeHound/pkg/telemetry/statsd" + "github.com/DataDog/KubeHound/pkg/telemetry/tag" ) type ProvidersFactoryConfig struct { @@ -81,6 +86,7 @@ func (p *ProvidersFactoryConfig) IngestBuildData(ctx context.Context, khCfg *con l := log.Logger(ctx) // Create the collector instance l.Info("Loading Kubernetes data collector client") + start := time.Now() collect, err := collector.ClientFactory(ctx, khCfg) if err != nil { return fmt.Errorf("collector client creation: %w", err) @@ -94,11 +100,23 @@ func (p *ProvidersFactoryConfig) IngestBuildData(ctx context.Context, khCfg *con if err != nil { return fmt.Errorf("raw data ingest: %w", err) } + // Metric for IngestData + _ = statsd.Gauge(ctx, metric.IngestionIngestDuration, float64(time.Since(start)), tag.GetDefaultTags(ctx), 1) + startBuild := time.Now() err = graph.BuildGraph(ctx, khCfg, p.StoreProvider, p.GraphProvider, p.CacheProvider) if err != nil { return err } + // Metric for BuildGraph + _ = statsd.Gauge(ctx, metric.IngestionBuildDuration, float64(time.Since(startBuild)), tag.GetDefaultTags(ctx), 1) + + // Metric for IngestBuildData + _ = statsd.Gauge(ctx, metric.IngestionRunDuration, float64(time.Since(start)), tag.GetDefaultTags(ctx), 1) + + text := fmt.Sprintf("KubeHound ingestion has been completed in %s", time.Since(start)) + _ = events.PushEvent(ctx, events.IngestFinished, text) + return nil } diff --git a/pkg/kubehound/storage/cache/memcache_provider.go b/pkg/kubehound/storage/cache/memcache_provider.go index 929a3b36d..40e0d43f6 100644 --- a/pkg/kubehound/storage/cache/memcache_provider.go +++ b/pkg/kubehound/storage/cache/memcache_provider.go @@ -61,10 +61,10 @@ func (m *MemCacheProvider) Get(ctx context.Context, key cachekey.CacheKey) *Cach data, ok := m.data[computeKey(key)] tagCacheKey := tag.GetBaseTagsWith(tag.CacheKey(key.Shard())) if !ok { - _ = statsd.Incr(metric.CacheMiss, tagCacheKey, 1) + _ = statsd.Incr(ctx, metric.CacheMiss, tagCacheKey, 1) log.Trace(ctx).Debugf("entry not found in cache: %s", computeKey(key)) } else { - _ = statsd.Incr(metric.CacheHit, tagCacheKey, 1) + _ = statsd.Incr(ctx, metric.CacheHit, tagCacheKey, 1) } return &CacheResult{ diff --git a/pkg/kubehound/storage/cache/memcache_writer.go b/pkg/kubehound/storage/cache/memcache_writer.go index 031aa5bee..ffbb3f4ff 100644 --- a/pkg/kubehound/storage/cache/memcache_writer.go +++ b/pkg/kubehound/storage/cache/memcache_writer.go @@ -22,7 +22,7 @@ func (m *MemCacheAsyncWriter) Queue(ctx context.Context, key cachekey.CacheKey, m.mu.Lock() defer m.mu.Unlock() tagCacheKey := tag.GetBaseTagsWith(tag.CacheKey(key.Shard())) - _ = statsd.Incr(metric.CacheWrite, tagCacheKey, 1) + _ = statsd.Incr(ctx, metric.CacheWrite, tagCacheKey, 1) keyId := computeKey(key) entry, ok := m.data[keyId] if ok { @@ -33,7 +33,7 @@ func (m *MemCacheAsyncWriter) Queue(ctx context.Context, key cachekey.CacheKey, if !m.opts.ExpectOverwrite { // if overwrite is expected (e.g fast tracking of existence regardless of value), suppress metrics and logs - _ = statsd.Incr(metric.CacheDuplicate, tagCacheKey, 1) + _ = statsd.Incr(ctx, metric.CacheDuplicate, tagCacheKey, 1) log.Trace(ctx).Warnf("overwriting cache entry key=%s old=%#v new=%#v", keyId, entry, value) } } diff --git a/pkg/kubehound/storage/graphdb/janusgraph_edge_writer.go b/pkg/kubehound/storage/graphdb/janusgraph_edge_writer.go index 93bf15088..4818a5e43 100644 --- a/pkg/kubehound/storage/graphdb/janusgraph_edge_writer.go +++ b/pkg/kubehound/storage/graphdb/janusgraph_edge_writer.go @@ -77,13 +77,13 @@ func (jgv *JanusGraphEdgeWriter) startBackgroundWriter(ctx context.Context) { return } - _ = statsd.Count(metric.BackgroundWriterCall, 1, jgv.tags, 1) + _ = statsd.Count(ctx, metric.BackgroundWriterCall, 1, jgv.tags, 1) err := jgv.batchWrite(ctx, data) if err != nil { log.Trace(ctx).Errorf("write data in background batch writer: %v", err) } - _ = statsd.Decr(metric.QueueSize, jgv.tags, 1) + _ = statsd.Decr(ctx, metric.QueueSize, jgv.tags, 1) case <-ctx.Done(): log.Trace(ctx).Info("Closed background janusgraph worker on context cancel") @@ -103,7 +103,7 @@ func (jgv *JanusGraphEdgeWriter) batchWrite(ctx context.Context, data []any) err defer jgv.writingInFlight.Done() datalen := len(data) - _ = statsd.Count(metric.EdgeWrite, int64(datalen), jgv.tags, 1) + _ = statsd.Count(ctx, metric.EdgeWrite, int64(datalen), jgv.tags, 1) log.Trace(ctx).Debugf("Batch write JanusGraphEdgeWriter with %d elements", datalen) atomic.AddInt32(&jgv.wcounter, int32(datalen)) @@ -139,7 +139,7 @@ func (jgv *JanusGraphEdgeWriter) Flush(ctx context.Context) error { } if len(jgv.inserts) != 0 { - _ = statsd.Incr(metric.FlushWriterCall, jgv.tags, 1) + _ = statsd.Incr(ctx, metric.FlushWriterCall, jgv.tags, 1) jgv.writingInFlight.Add(1) err = jgv.batchWrite(ctx, jgv.inserts) @@ -175,7 +175,7 @@ func (jgv *JanusGraphEdgeWriter) Queue(ctx context.Context, v any) error { jgv.writingInFlight.Add(1) jgv.consumerChan <- copied - _ = statsd.Incr(metric.QueueSize, jgv.tags, 1) + _ = statsd.Incr(ctx, metric.QueueSize, jgv.tags, 1) // cleanup the ops array after we have copied it to the channel jgv.inserts = nil diff --git a/pkg/kubehound/storage/graphdb/janusgraph_vertex_writer.go b/pkg/kubehound/storage/graphdb/janusgraph_vertex_writer.go index 55a92ca05..ec7c8c9ad 100644 --- a/pkg/kubehound/storage/graphdb/janusgraph_vertex_writer.go +++ b/pkg/kubehound/storage/graphdb/janusgraph_vertex_writer.go @@ -81,13 +81,13 @@ func (jgv *JanusGraphVertexWriter) startBackgroundWriter(ctx context.Context) { return } - _ = statsd.Count(metric.BackgroundWriterCall, 1, jgv.tags, 1) + _ = statsd.Count(ctx, metric.BackgroundWriterCall, 1, jgv.tags, 1) err := jgv.batchWrite(ctx, data) if err != nil { log.Trace(ctx).Errorf("Write data in background batch writer: %v", err) } - _ = statsd.Decr(metric.QueueSize, jgv.tags, 1) + _ = statsd.Decr(ctx, metric.QueueSize, jgv.tags, 1) case <-ctx.Done(): log.Trace(ctx).Info("Closed background janusgraph worker on context cancel") @@ -130,7 +130,7 @@ func (jgv *JanusGraphVertexWriter) batchWrite(ctx context.Context, data []any) e defer jgv.writingInFlight.Done() datalen := len(data) - _ = statsd.Count(metric.VertexWrite, int64(datalen), jgv.tags, 1) + _ = statsd.Count(ctx, metric.VertexWrite, int64(datalen), jgv.tags, 1) log.Trace(ctx).Debugf("Batch write JanusGraphVertexWriter with %d elements", datalen) atomic.AddInt32(&jgv.wcounter, int32(datalen)) @@ -174,7 +174,7 @@ func (jgv *JanusGraphVertexWriter) Flush(ctx context.Context) error { } if len(jgv.inserts) != 0 { - _ = statsd.Incr(metric.FlushWriterCall, jgv.tags, 1) + _ = statsd.Incr(ctx, metric.FlushWriterCall, jgv.tags, 1) jgv.writingInFlight.Add(1) err = jgv.batchWrite(ctx, jgv.inserts) @@ -215,7 +215,7 @@ func (jgv *JanusGraphVertexWriter) Queue(ctx context.Context, v any) error { jgv.writingInFlight.Add(1) jgv.consumerChan <- copied - _ = statsd.Incr(metric.QueueSize, jgv.tags, 1) + _ = statsd.Incr(ctx, metric.QueueSize, jgv.tags, 1) // cleanup the ops array after we have copied it to the channel jgv.inserts = nil diff --git a/pkg/kubehound/storage/storedb/mongo_writer.go b/pkg/kubehound/storage/storedb/mongo_writer.go index ad606852b..dbf0860ce 100644 --- a/pkg/kubehound/storage/storedb/mongo_writer.go +++ b/pkg/kubehound/storage/storedb/mongo_writer.go @@ -67,13 +67,13 @@ func (maw *MongoAsyncWriter) startBackgroundWriter(ctx context.Context) { return } - _ = statsd.Count(metric.BackgroundWriterCall, 1, maw.tags, 1) + _ = statsd.Count(ctx, metric.BackgroundWriterCall, 1, maw.tags, 1) err := maw.batchWrite(ctx, data) if err != nil { log.Trace(ctx).Errorf("write data in background batch writer: %v", err) } - _ = statsd.Decr(metric.QueueSize, maw.tags, 1) + _ = statsd.Decr(ctx, metric.QueueSize, maw.tags, 1) case <-ctx.Done(): log.Trace(ctx).Debug("Closed background mongodb worker") @@ -91,7 +91,7 @@ func (maw *MongoAsyncWriter) batchWrite(ctx context.Context, ops []mongo.WriteMo defer func() { span.Finish(tracer.WithError(err)) }() defer maw.writingInFlight.Done() - _ = statsd.Count(metric.ObjectWrite, int64(len(ops)), maw.tags, 1) + _ = statsd.Count(ctx, metric.ObjectWrite, int64(len(ops)), maw.tags, 1) bulkWriteOpts := options.BulkWrite().SetOrdered(false) _, err = maw.dbWriter.BulkWrite(ctx, ops, bulkWriteOpts) @@ -114,7 +114,7 @@ func (maw *MongoAsyncWriter) Queue(ctx context.Context, model any) error { maw.writingInFlight.Add(1) maw.consumerChan <- copied - _ = statsd.Incr(metric.QueueSize, maw.tags, 1) + _ = statsd.Incr(ctx, metric.QueueSize, maw.tags, 1) // cleanup the ops array after we have copied it to the channel maw.ops = nil diff --git a/pkg/telemetry/events/events.go b/pkg/telemetry/events/events.go index f4e2b363b..5e008e3fa 100644 --- a/pkg/telemetry/events/events.go +++ b/pkg/telemetry/events/events.go @@ -1,19 +1,95 @@ package events import ( + "context" + "fmt" + + "github.com/DataDog/KubeHound/pkg/telemetry/log" kstatsd "github.com/DataDog/KubeHound/pkg/telemetry/statsd" + "github.com/DataDog/KubeHound/pkg/telemetry/tag" "github.com/DataDog/datadog-go/v5/statsd" ) const ( - DumperRun = "kubehound.dumper.run" - DumperStop = "kubehound.dumper.stop" + IngestSkip = iota + IngestStarted + IngestFinished + IngestorInit + IngestorFailed + DumpStarted + DumpFinished +) + +const ( + EventActionFail = "fail" + EventActionInit = "init" + EventActionStart = "start" + EventActionSkip = "skip" + EventActionFinish = "finish" ) -func PushEvent(title string, text string, tags []string) { - _ = kstatsd.Event(&statsd.Event{ - Title: title, - Text: text, - Tags: tags, +type EventAction int + +type EventActionDetails struct { + Title string + Text string + Level statsd.EventAlertType + Action string +} + +// Could also be a format stirng template in this case, if needed? +var map2msg = map[EventAction]EventActionDetails{ + IngestorFailed: {Title: "Ingestor/grpc endpoint init failed", Level: statsd.Error, Action: EventActionFail}, + IngestorInit: {Title: "Ingestor/grpc endpoint initiated", Level: statsd.Info, Action: EventActionInit}, + IngestStarted: {Title: "Ingestion started", Level: statsd.Info, Action: EventActionStart}, + IngestSkip: {Title: "Ingestion skipped", Level: statsd.Info, Action: EventActionSkip}, + IngestFinished: {Title: "Ingestion finished", Level: statsd.Info, Action: EventActionFinish}, + + DumpStarted: {Title: "Dump started", Level: statsd.Info, Action: EventActionStart}, + DumpFinished: {Title: "Dump finished", Level: statsd.Info, Action: EventActionFinish}, +} + +func (ea EventAction) Tags(ctx context.Context) []string { + tags := tag.GetDefaultTags(ctx) + tags = append(tags, fmt.Sprintf("%s:%s", tag.ActionTypeTag, map2msg[ea].Action)) + + return tags +} + +func (ea EventAction) Level() statsd.EventAlertType { + return map2msg[ea].Level +} + +func (ea EventAction) Title(ctx context.Context) string { + title, _ := getTitleTextMsg(ctx, map2msg[ea].Title) + + return title +} + +func (ea EventAction) DefaultMessage(ctx context.Context) string { + _, msg := getTitleTextMsg(ctx, map2msg[ea].Title) + + return msg +} + +func getTitleTextMsg(ctx context.Context, actionMsg string) (string, string) { + cluster := log.GetClusterFromContext(ctx) + runId := log.GetRunIDFromContext(ctx) + title := fmt.Sprintf("%s for %s", actionMsg, cluster) + text := fmt.Sprintf("%s for %s with run_id %s", actionMsg, cluster, runId) + + return title, text +} + +func PushEvent(ctx context.Context, action EventAction, text string) error { + if text == "" { + text = action.DefaultMessage(ctx) + } + + return kstatsd.Event(&statsd.Event{ + Title: action.Title(ctx), + Text: text, + Tags: action.Tags(ctx), + AlertType: action.Level(), }) } diff --git a/pkg/telemetry/log/fields.go b/pkg/telemetry/log/fields.go index 01915c3f7..07fec0f4d 100644 --- a/pkg/telemetry/log/fields.go +++ b/pkg/telemetry/log/fields.go @@ -47,6 +47,18 @@ func convertField(value any) string { return val } +func GetRunIDFromContext(ctx context.Context) string { + return convertField(ctx.Value(ContextFieldRunID)) +} + +func GetClusterFromContext(ctx context.Context) string { + return convertField(ctx.Value(ContextFieldCluster)) +} + +func GetComponentFromContext(ctx context.Context) string { + return convertField(ctx.Value(ContextFieldComponent)) +} + func SpanSetDefaultField(ctx context.Context, span ddtrace.Span) { runID := convertField(ctx.Value(ContextFieldRunID)) if runID != "" { diff --git a/pkg/telemetry/metric/metrics.go b/pkg/telemetry/metric/metrics.go index e2040bcb6..afb927704 100644 --- a/pkg/telemetry/metric/metrics.go +++ b/pkg/telemetry/metric/metrics.go @@ -37,3 +37,10 @@ var ( CacheWrite = "kubehound.cache.write" CacheDuplicate = "kubehound.cache.duplicate" ) + +// Ingestion metrics +const ( + IngestionRunDuration = "kubehound.ingestion.run.duration" + IngestionBuildDuration = "kubehound.ingestion.build.duration" + IngestionIngestDuration = "kubehound.ingestion.ingest.duration" +) diff --git a/pkg/telemetry/span/spans.go b/pkg/telemetry/span/spans.go index 3d1537e8d..bd4ec114d 100644 --- a/pkg/telemetry/span/spans.go +++ b/pkg/telemetry/span/spans.go @@ -64,15 +64,6 @@ const ( BuildEdge = "kubehound.graph.builder.edge" ) -// to avoid the following lint error -// should not use built-in type string as key for value; define your own type to avoid collisions (SA1029) -type contextKey int - -const ( - ContextLogFieldClusterName contextKey = iota - ContextLogFieldRunID -) - func convertTag(value any) string { val, err := value.(string) if !err { @@ -82,6 +73,13 @@ func convertTag(value any) string { return val } +func StartSpanFromContext(runCtx context.Context, operationName string, opts ...tracer.StartSpanOption) (tracer.Span, context.Context) { + spanJob, runCtx := tracer.StartSpanFromContext(runCtx, operationName, opts...) + spanIngestRunSetDefaultTag(runCtx, spanJob) + + return spanJob, runCtx +} + func SpanRunFromContext(runCtx context.Context, spanName string) (ddtrace.Span, context.Context) { spanJob, runCtx := tracer.StartSpanFromContext(runCtx, spanName, tracer.ResourceName(convertTag(runCtx.Value(log.ContextFieldCluster))), tracer.Measured()) spanIngestRunSetDefaultTag(runCtx, spanJob) diff --git a/pkg/telemetry/statsd/statsd.go b/pkg/telemetry/statsd/statsd.go index 1f6b02e3a..0114dfcf2 100644 --- a/pkg/telemetry/statsd/statsd.go +++ b/pkg/telemetry/statsd/statsd.go @@ -50,46 +50,51 @@ func Setup(ctx context.Context, cfg *config.KubehoundConfig) error { } // Count tracks how many times something happened per second. -func Count(name string, value int64, tags []string, rate float64) error { +func Count(ctx context.Context, name string, value int64, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Count(name, value, tags, rate) } // Gauge measures the value of a metric at a particular time. -func Gauge(name string, value float64, tags []string, rate float64) error { +func Gauge(ctx context.Context, name string, value float64, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Gauge(name, value, tags, rate) } // Incr is just Count of 1 -func Incr(name string, tags []string, rate float64) error { +func Incr(ctx context.Context, name string, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Incr(name, tags, rate) } // Decr is just Count of -1 -func Decr(name string, tags []string, rate float64) error { +func Decr(ctx context.Context, name string, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Decr(name, tags, rate) } // Histogram tracks the statistical distribution of a set of values. -func Histogram(name string, value float64, tags []string, rate float64) error { +func Histogram(ctx context.Context, name string, value float64, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Histogram(name, value, tags, rate) } @@ -113,28 +118,31 @@ func SimpleEvent(title string, text string) error { } // Set counts the number of unique elements in a group. -func Set(name string, value string, tags []string, rate float64) error { +func Set(ctx context.Context, name string, value string, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Set(name, value, tags, rate) } // Timing sends timing information, it is an alias for TimeInMilliseconds -func Timing(name string, value time.Duration, tags []string, rate float64) error { +func Timing(ctx context.Context, name string, value time.Duration, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Timing(name, value, tags, rate) } // TimingDist sends dt in milliseconds as a distribution (p50-p99) -func TimingDist(name string, dt time.Duration, tags []string, rate float64) error { +func TimingDist(ctx context.Context, name string, dt time.Duration, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) const secToMillis = 1000 @@ -142,19 +150,21 @@ func TimingDist(name string, dt time.Duration, tags []string, rate float64) erro } // TimeInMilliseconds sends timing information in milliseconds. -func TimeInMilliseconds(name string, value float64, tags []string, rate float64) error { +func TimeInMilliseconds(ctx context.Context, name string, value float64, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.TimeInMilliseconds(name, value, tags, rate) } // Distribution tracks accurate global percentiles of a set of values. -func Distribution(name string, value float64, tags []string, rate float64) error { +func Distribution(ctx context.Context, name string, value float64, tags []string, rate float64) error { if statsdClient == nil { return nil } + tags = append(tags, tag.GetDefaultTags(ctx)...) return statsdClient.Distribution(name, value, tags, rate) } diff --git a/pkg/telemetry/tag/tags.go b/pkg/telemetry/tag/tags.go index f9340a60c..3e3b6693d 100644 --- a/pkg/telemetry/tag/tags.go +++ b/pkg/telemetry/tag/tags.go @@ -1,20 +1,20 @@ package tag -import "sync" +import ( + "context" + "sync" + + "github.com/DataDog/KubeHound/pkg/telemetry/log" +) const ( ActionTypeTag = "action" CollectorTag = "collector" - CollectorClusterTag = "cluster" DumperS3BucketTag = "s3_bucket" DumperS3keyTag = "s3_key" - DumperFilePathTag = "file_path" DumperWorkerNumberTag = "worker_number" DumperWriterTypeTag = "writer_type" - EntityTag = "entity" WaitTag = "wait" - RunIdTag = "run_id" - IngestionRunIdTag = "ingestion_run_id" LabelTag = "label" CollectionTag = "collection" BuilderTag = "builder" @@ -23,6 +23,14 @@ const ( EdgeTypeTag = "edge_type" ) +var ( + RunIdTag = log.FieldRunIDKey + CollectorClusterTag = log.FieldClusterKey + DumperFilePathTag = log.FieldPathKey + EntityTag = log.FieldEntityKey + CompenentTag = log.FieldComponentKey +) + const ( StorageJanusGraph = "janusgraph" StorageMongoDB = "mongodb" @@ -83,10 +91,6 @@ func RunID(uuid string) string { return MakeTag(RunIdTag, uuid) } -func IngestionRunID(uuid string) string { - return MakeTag(IngestionRunIdTag, uuid) -} - func Collector(collector string) string { return MakeTag(CollectorTag, collector) } @@ -134,3 +138,27 @@ func S3Bucket(bucket string) string { func S3Key(key string) string { return MakeTag(DumperS3keyTag, key) } + +func ComponentName(component string) string { + return MakeTag(CompenentTag, component) +} + +func GetDefaultTags(ctx context.Context) []string { + defaultTags := []string{} + runID := log.GetRunIDFromContext(ctx) + if runID != "" { + defaultTags = append(defaultTags, RunID(runID)) + } + + cluster := log.GetClusterFromContext(ctx) + if cluster != "" { + defaultTags = append(defaultTags, ClusterName(cluster)) + } + + component := log.GetComponentFromContext(ctx) + if component != "" { + defaultTags = append(defaultTags, ComponentName(component)) + } + + return defaultTags +}