only-one-copy-in-graph-per-cluster

pkg/ingestor/api/api.go

@@ -56,10 +56,13 @@ func NewIngestorAPI(cfg *config.KubehoundConfig, puller puller.DataPuller, notif
 	}
 }
 
+func (g *IngestorAPI) Close(ctx context.Context) {
+	g.providers.Close(ctx)
+}
+
 // RehydrateLatest is just a GRPC wrapper around the Ingest method from the API package
 func (g *IngestorAPI) RehydrateLatest(ctx context.Context) ([]*grpc.IngestedCluster, error) {
 	l := log.Logger(ctx)
-	l.Error("id123")
 	// first level key are cluster names
 	directories, errRet := g.puller.ListFiles(ctx, "", false)
 	if errRet != nil {

pkg/kubehound/core/core_grpc_api.go

@@ -15,7 +15,7 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
 )
 
-func CoreGrpcApi(ctx context.Context, khCfg *config.KubehoundConfig) error {
+func initCoreGrpcApi(ctx context.Context, khCfg *config.KubehoundConfig) (*api.IngestorAPI, error) {
 	l := log.Logger(ctx)
 	l.Info("Starting KubeHound Distributed Ingestor Service")
 	span, ctx := span.SpanRunFromContext(ctx, span.IngestorLaunch)
@@ -28,24 +28,36 @@ func CoreGrpcApi(ctx context.Context, khCfg *config.KubehoundConfig) error {
 	l.Info("Initializing providers (graph, cache, store)")
 	p, err := providers.NewProvidersFactoryConfig(ctx, khCfg)
 	if err != nil {
-		return fmt.Errorf("factory config creation: %w", err)
+		return nil, fmt.Errorf("factory config creation: %w", err)
 	}
-	defer p.Close(ctx)
 
 	l.Info("Creating Blob Storage provider")
 	puller, err := blob.NewBlobStorage(khCfg, khCfg.Ingestor.Blob)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	l.Info("Creating Noop Notifier")
 	noopNotifier := noop.NewNoopNotifier()
 
 	l.Info("Creating Ingestor API")
-	ingestorApi := api.NewIngestorAPI(khCfg, puller, noopNotifier, p)
 
+	return api.NewIngestorAPI(khCfg, puller, noopNotifier, p), nil
+}
+
+func CoreGrpcApi(ctx context.Context, khCfg *config.KubehoundConfig) error {
+	ingestorApi, _ := initCoreGrpcApi(ctx, khCfg)
+	defer ingestorApi.Close(ctx)
+	// TODO: need to merge fix-observability branch to get the following events
+	// if err != nil {
+	// 	events.PushEventIngestorFailed(ctx)
+	// 	return err
+	// }
+	// events.PushEventIngestorInit(ctx)
+
+	l := log.Logger(ctx)
 	l.Info("Starting Ingestor API")
-	err = grpc.Listen(ctx, ingestorApi)
+	err := grpc.Listen(ctx, ingestorApi)
 	if err != nil {
 		return err
 	}

pkg/kubehound/storage/graphdb/janusgraph_provider.go

@@ -145,3 +145,29 @@ func (jgp *JanusGraphProvider) Close(ctx context.Context) error {
 
 	return nil
 }
+
+// Raw returns a handle to the underlying provider to allow implementation specific operations e.g graph queries.
+func (jgp *JanusGraphProvider) Clean(ctx context.Context, cluster string) error {
+	l := log.Trace(ctx)
+	l.Infof("Cleaning cluster", log.FieldClusterKey, cluster)
+	g := gremlin.Traversal_().WithRemote(jgp.drc)
+	tx := g.Tx()
+	defer tx.Close()
+
+	gtx, err := tx.Begin()
+	if err != nil {
+		return err
+	}
+
+	err = <-gtx.V().Has("cluster", cluster).Drop().Iterate()
+	if err != nil {
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/kubehound/storage/graphdb/provider.go

@@ -35,6 +35,9 @@ type Provider interface {
 	// Raw returns a handle to the underlying provider to allow implementation specific operations e.g graph queries.
 	Raw() any
 
+	// Droping all assets from the graph database from a cluster name
+	Clean(ctx context.Context, cluster string) error
+
 	// VertexWriter creates a new AsyncVertexWriter instance to enable asynchronous bulk inserts of vertices.
 	VertexWriter(ctx context.Context, v vertex.Builder, c cache.CacheProvider, opts ...WriterOption) (AsyncVertexWriter, error)