Bed 4069

cmd/api/src/daemons/datapipe/analysis.go

@@ -18,12 +18,13 @@ package datapipe
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"github.com/specterops/bloodhound/log"
 
 	"github.com/specterops/bloodhound/analysis"
 	adAnalysis "github.com/specterops/bloodhound/analysis/ad"
 	"github.com/specterops/bloodhound/dawgs/graph"
-	"github.com/specterops/bloodhound/errors"
 	"github.com/specterops/bloodhound/src/analysis/ad"
 	"github.com/specterops/bloodhound/src/analysis/azure"
 	"github.com/specterops/bloodhound/src/config"
@@ -33,57 +34,81 @@ import (
 	"github.com/specterops/bloodhound/src/services/dataquality"
 )
 
+var (
+	ErrAnalysisFailed             = errors.New("analysis failed")
+	ErrAnalysisPartiallyCompleted = errors.New("analysis partially completed")
+)
+
 func RunAnalysisOperations(ctx context.Context, db database.Database, graphDB graph.Database, _ config.Configuration) error {
-	var (
-		collector = &errors.ErrorCollector{}
-	)
+	var collectedErrors error
 
 	if err := adAnalysis.FixWellKnownNodeTypes(ctx, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("fix well known node types failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("fix well known node types failed: %w", err))
 	}
 
 	if err := adAnalysis.RunDomainAssociations(ctx, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("domain association and pruning failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("domain association and pruning failed: %w", err))
 	}
 
 	if err := adAnalysis.LinkWellKnownGroups(ctx, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("well known group linking failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("well known group linking failed: %w", err))
 	}
 
 	if err := updateAssetGroupIsolationTags(ctx, db, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("asset group isolation tagging failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("asset group isolation tagging failed: %w", err))
 	}
 
 	if err := TagActiveDirectoryTierZero(ctx, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("active directory tier zero tagging failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("active directory tier zero tagging failed: %w", err))
 	}
 
 	if err := ParallelTagAzureTierZero(ctx, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("azure tier zero tagging failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("azure tier zero tagging failed: %w", err))
 	}
 
+	var (
+		adFailed          = false
+		azureFailed       = false
+		agiFailed         = false
+		dataQualityFailed = false
+	)
+
 	// TODO: Cleanup #ADCSFeatureFlag after full launch.
 	if adcsFlag, err := db.GetFlagByKey(appcfg.FeatureAdcs); err != nil {
-		collector.Collect(fmt.Errorf("error retrieving ADCS feature flag: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("error retrieving ADCS feature flag: %w", err))
 	} else if stats, err := ad.Post(ctx, graphDB, adcsFlag.Enabled); err != nil {
-		collector.Collect(fmt.Errorf("error during ad post: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("error during ad post: %w", err))
+		adFailed = true
 	} else {
 		stats.LogStats()
 	}
 
 	if stats, err := azure.Post(ctx, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("error during azure post: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("error during azure post: %w", err))
+		azureFailed = true
 	} else {
 		stats.LogStats()
 	}
 
 	if err := agi.RunAssetGroupIsolationCollections(ctx, db, graphDB, analysis.GetNodeKindDisplayLabel); err != nil {
-		collector.Collect(fmt.Errorf("asset group isolation collection failed: %w", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("asset group isolation collection failed: %w", err))
+		agiFailed = true
 	}
 
 	if err := dataquality.SaveDataQuality(ctx, db, graphDB); err != nil {
-		collector.Collect(fmt.Errorf("error saving data quality stat: %v", err))
+		collectedErrors = errors.Join(collectedErrors, fmt.Errorf("error saving data quality stat: %v", err))
+		dataQualityFailed = true
+	}
+
+	if collectedErrors != nil {
+		log.Errorf("Analysis errors encountered: %v", collectedErrors)
+	}
+
+	if adFailed && azureFailed && agiFailed && dataQualityFailed {
+		return ErrAnalysisFailed
+	} else if adFailed || azureFailed || agiFailed || dataQualityFailed {
+		return ErrAnalysisPartiallyCompleted
 	}
 
-	return collector.Return()
+	return nil
 }

cmd/api/src/daemons/datapipe/datapipe.go

@@ -19,6 +19,7 @@ package datapipe
 
 import (
 	"context"
+	"errors"
 	"github.com/specterops/bloodhound/src/bootstrap"
 	"os"
 	"path/filepath"
@@ -107,10 +108,13 @@ func (s *Daemon) analyze() {
 	log.LogAndMeasure(log.LevelInfo, "Graph Analysis")()
 
 	if err := RunAnalysisOperations(s.ctx, s.db, s.graphdb, s.cfg); err != nil {
-		log.Errorf("Graph analysis failed: %v", err)
-		FailAnalyzedFileUploadJobs(s.ctx, s.db)
-
-		s.status.Update(model.DatapipeStatusIdle, false)
+		if errors.Is(err, ErrAnalysisFailed) {
+			FailAnalyzedFileUploadJobs(s.ctx, s.db)
+			s.status.Update(model.DatapipeStatusIdle, false)
+		} else if errors.Is(err, ErrAnalysisPartiallyCompleted) {
+			PartialCompleteFileUploadJobs(s.ctx, s.db)
+			s.status.Update(model.DatapipeStatusIdle, true)
+		}
 	} else {
 		CompleteAnalyzedFileUploadJobs(s.ctx, s.db)
 

cmd/api/src/daemons/datapipe/jobs.go

@@ -54,6 +54,24 @@ func FailAnalyzedFileUploadJobs(ctx context.Context, db database.Database) {
 	}
 }
 
+func PartialCompleteFileUploadJobs(ctx context.Context, db database.Database) {
+	// Because our database interfaces do not yet accept contexts this is a best-effort check to ensure that we do not
+	// commit state transitions when we are shutting down.
+	if ctx.Err() != nil {
+		return
+	}
+
+	if fileUploadJobsUnderAnalysis, err := db.GetFileUploadJobsWithStatus(model.JobStatusAnalyzing); err != nil {
+		log.Errorf("Failed to load file upload jobs under analysis: %v", err)
+	} else {
+		for _, job := range fileUploadJobsUnderAnalysis {
+			if err := fileupload.UpdateFileUploadJobStatus(db, job, model.JobStatusPartiallyComplete, "Analysis Partially Completed"); err != nil {
+				log.Errorf("Failed updating file upload job %d to failed status: %v", job.ID, err)
+			}
+		}
+	}
+}
+
 func CompleteAnalyzedFileUploadJobs(ctx context.Context, db database.Database) {
 	// Because our database interfaces do not yet accept contexts this is a best-effort check to ensure that we do not
 	// commit state transitions when we are shutting down.

cmd/api/src/model/jobs.go

@@ -117,6 +117,7 @@ const (
 	JobStatusFailed    JobStatus = 5
 	JobStatusIngesting JobStatus = 6
 	JobStatusAnalyzing JobStatus = 7
+	JobStatusPartiallyComplete JobStatus = 8
 )
 
 func allJobStatuses() []JobStatus {
@@ -130,6 +131,7 @@ func allJobStatuses() []JobStatus {
 		JobStatusFailed,
 		JobStatusIngesting,
 		JobStatusAnalyzing,
+		JobStatusPartiallyComplete,
 	}
 }
 
@@ -170,6 +172,9 @@ func (s JobStatus) String() string {
 
 	case JobStatusAnalyzing:
 		return "ANALYZING"
+
+	case JobStatusPartiallyComplete:
+		return "PARTIALLYCOMPLETE"
 
 	default:
 		return "INVALIDSTATUS"

packages/javascript/bh-shared-ui/src/components/FinishedIngestLog/types.ts

@@ -35,6 +35,7 @@ export enum FileUploadJobStatus {
     FAILED = 5,
     INGESTING = 6,
     ANALYZING = 7,
+    PARTIALLY_COMPLETE = 8,
 }
 
 export const FileUploadJobStatusToString: Record<FileUploadJobStatus, string> = {
@@ -47,4 +48,5 @@ export const FileUploadJobStatusToString: Record<FileUploadJobStatus, string> =
     5: 'Failed',
     6: 'Ingesting',
     7: 'Analyzing',
+    8: 'Partially Complete',
 };