Start using new eval history table

This PR wires the eval history service into the evaluation engine. It is
gated behind a feature flag while we experiment with it. A follow on PR
will add logging for remediation and alerting.

Relates to #3556

database/query/eval_history.sql

@@ -65,5 +65,5 @@ INSERT INTO latest_evaluation_statuses(
     $1,
     $2
 )
-ON CONFLICT (rule_entity_id, evaluation_history_id) DO UPDATE
+ON CONFLICT (rule_entity_id) DO UPDATE
 SET evaluation_history_id = $2;

internal/controlplane/server.go

@@ -141,6 +141,7 @@ func NewServer(
 	sessionService session.ProviderSessionService,
 	projectDeleter projects.ProjectDeleter,
 	projectCreator projects.ProjectCreator,
+	featureFlagClient *openfeature.Client,
 ) *Server {
 	return &Server{
 		store:               store,
@@ -152,7 +153,7 @@ func NewServer(
 		profiles:            profileService,
 		ruleTypes:           ruleService,
 		providerStore:       providerStore,
-		featureFlags:        openfeature.NewClient(cfg.Flags.AppName),
+		featureFlags:        featureFlagClient,
 		ghClient:            &ghprov.ClientServiceImplementation{},
 		providerManager:     providerManager,
 		providerAuthManager: providerAuthManager,

internal/engine/eval_status.go

@@ -28,6 +28,7 @@ import (
 	evalerrors "github.com/stacklok/minder/internal/engine/errors"
 	engif "github.com/stacklok/minder/internal/engine/interfaces"
 	ent "github.com/stacklok/minder/internal/entities"
+	"github.com/stacklok/minder/internal/flags"
 	pb "github.com/stacklok/minder/pkg/api/protobuf/go/minder/v1"
 )
 
@@ -126,8 +127,21 @@ func (e *Executor) createOrUpdateEvalStatus(
 		return nil
 	}
 
+	// Get rule instance ID
+	// TODO: we should use the rule instance table in the evaluation process
+	// it should not be necessary to query it here.
+	ruleID, err := e.querier.GetIDByProfileEntityName(ctx, db.GetIDByProfileEntityNameParams{
+		ProfileID:  params.ProfileID,
+		EntityType: params.EntityType,
+		Name:       params.Rule.Name,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to retrieve rule instance ID from DB: %w", err)
+	}
+
 	// Upsert evaluation
-	id, err := e.querier.UpsertRuleEvaluations(ctx, db.UpsertRuleEvaluationsParams{
+	// TODO: replace this table with the evaluation statuses table
+	evalID, err := e.querier.UpsertRuleEvaluations(ctx, db.UpsertRuleEvaluationsParams{
 		ProfileID:     params.ProfileID,
 		RepositoryID:  params.RepoID,
 		ArtifactID:    params.ArtifactID,
@@ -136,7 +150,6 @@ func (e *Executor) createOrUpdateEvalStatus(
 		PullRequestID: params.PullRequestID,
 		RuleName:      params.Rule.Name,
 	})
-
 	if err != nil {
 		logger.Err(err).Msg("error upserting rule evaluation")
 		return err
@@ -150,7 +163,7 @@ func (e *Executor) createOrUpdateEvalStatus(
 	status := evalerrors.ErrorAsEvalStatus(params.GetEvalErr())
 	e.metrics.CountEvalStatus(ctx, status, params.ProfileID, params.ProjectID, entityID, entityType)
 	_, err = e.querier.UpsertRuleDetailsEval(ctx, db.UpsertRuleDetailsEvalParams{
-		RuleEvalID: id,
+		RuleEvalID: evalID,
 		Status:     evalerrors.ErrorAsEvalStatus(params.GetEvalErr()),
 		Details:    evalerrors.ErrorAsEvalDetails(params.GetEvalErr()),
 	})
@@ -161,9 +174,12 @@ func (e *Executor) createOrUpdateEvalStatus(
 	}
 
 	// Upsert remediation details
+	remediationStatus := evalerrors.ErrorAsRemediationStatus(params.GetActionsErr().RemediateErr)
+	e.metrics.CountRemediationStatus(ctx, remediationStatus, evalID, params.ProjectID)
+
 	_, err = e.querier.UpsertRuleDetailsRemediate(ctx, db.UpsertRuleDetailsRemediateParams{
-		RuleEvalID: id,
-		Status:     evalerrors.ErrorAsRemediationStatus(params.GetActionsErr().RemediateErr),
+		RuleEvalID: evalID,
+		Status:     remediationStatus,
 		Details:    errorAsActionDetails(params.GetActionsErr().RemediateErr),
 		Metadata:   params.GetActionsErr().RemediateMeta,
 	})
@@ -172,16 +188,37 @@ func (e *Executor) createOrUpdateEvalStatus(
 	}
 
 	// Upsert alert details
+	alertStatus := evalerrors.ErrorAsAlertStatus(params.GetActionsErr().AlertErr)
+	e.metrics.CountAlertStatus(ctx, alertStatus, evalID, params.ProjectID)
+
 	_, err = e.querier.UpsertRuleDetailsAlert(ctx, db.UpsertRuleDetailsAlertParams{
-		RuleEvalID: id,
-		Status:     evalerrors.ErrorAsAlertStatus(params.GetActionsErr().AlertErr),
+		RuleEvalID: evalID,
+		Status:     alertStatus,
 		Details:    errorAsActionDetails(params.GetActionsErr().AlertErr),
 		Metadata:   params.GetActionsErr().AlertMeta,
 	})
 	if err != nil {
 		logger.Err(err).Msg("error upserting rule alert details")
 	}
 
+	if flags.Bool(ctx, e.featureFlags, flags.EvalHistory) {
+		// Log in the evaluation history tables
+		_, err = db.WithTransaction(e.querier, func(qtx db.ExtendQuerier) (uuid.UUID, error) {
+			return e.historyService.StoreEvaluationStatus(
+				ctx,
+				qtx,
+				ruleID,
+				params.EntityType,
+				entityID,
+				params.GetEvalErr(),
+			)
+		})
+		if err != nil {
+			logger.Err(err).Msg("error logging evaluation status")
+			return err
+		}
+	}
+
 	return err
 }
 

internal/engine/executor.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/google/uuid"
+	"github.com/open-feature/go-sdk/openfeature"
 	"github.com/rs/zerolog"
 
 	"github.com/stacklok/minder/internal/db"
@@ -33,6 +34,7 @@ import (
 	"github.com/stacklok/minder/internal/engine/ingestcache"
 	engif "github.com/stacklok/minder/internal/engine/interfaces"
 	"github.com/stacklok/minder/internal/events"
+	"github.com/stacklok/minder/internal/history"
 	minderlogger "github.com/stacklok/minder/internal/logger"
 	"github.com/stacklok/minder/internal/profiles"
 	"github.com/stacklok/minder/internal/providers/manager"
@@ -61,6 +63,8 @@ type Executor struct {
 	terminationcontext context.Context
 	providerManager    manager.ProviderManager
 	metrics            *ExecutorMetrics
+	historyService     history.EvaluationHistoryService
+	featureFlags       openfeature.IClient
 }
 
 // NewExecutor creates a new executor
@@ -71,6 +75,8 @@ func NewExecutor(
 	providerManager manager.ProviderManager,
 	handlerMiddleware []message.HandlerMiddleware,
 	metrics *ExecutorMetrics,
+	historyService history.EvaluationHistoryService,
+	featureFlags openfeature.IClient,
 ) *Executor {
 	return &Executor{
 		querier:                querier,
@@ -80,6 +86,8 @@ func NewExecutor(
 		handlerMiddleware:      handlerMiddleware,
 		providerManager:        providerManager,
 		metrics:                metrics,
+		historyService:         historyService,
+		featureFlags:           featureFlags,
 	}
 }
 
@@ -390,7 +398,8 @@ func (e *Executor) releaseLockAndFlush(
 func logEval(
 	ctx context.Context,
 	inf *entities.EntityInfoWrapper,
-	params *engif.EvalStatusParams) {
+	params *engif.EvalStatusParams,
+) {
 	evalLog := params.DecorateLogger(
 		zerolog.Ctx(ctx).With().
 			Str("eval_status", string(evalerrors.ErrorAsEvalStatus(params.GetEvalErr()))).

internal/engine/executor_test.go

@@ -42,6 +42,8 @@ import (
 	"github.com/stacklok/minder/internal/engine/actions/remediate"
 	"github.com/stacklok/minder/internal/engine/entities"
 	"github.com/stacklok/minder/internal/events"
+	"github.com/stacklok/minder/internal/flags"
+	mock_history "github.com/stacklok/minder/internal/history/mock"
 	"github.com/stacklok/minder/internal/logger"
 	"github.com/stacklok/minder/internal/metrics/meters"
 	"github.com/stacklok/minder/internal/providers"
@@ -255,6 +257,15 @@ default allow = true`,
 			Details:    "",
 		}).Return(ruleEvalDetailsId, nil)
 
+	ruleInstanceID := uuid.New()
+	mockStore.EXPECT().
+		GetIDByProfileEntityName(gomock.Any(), db.GetIDByProfileEntityNameParams{
+			ProfileID:  profileID,
+			EntityType: db.EntitiesRepository,
+			Name:       passthroughRuleType,
+		}).
+		Return(ruleInstanceID, nil)
+
 	// Mock upserting remediate status
 	ruleEvalRemediationId := uuid.New()
 	mockStore.EXPECT().
@@ -350,6 +361,7 @@ default allow = true`,
 
 	execMetrics, err := engine.NewExecutorMetrics(&meters.NoopMeterFactory{})
 	require.NoError(t, err)
+	historyService := mock_history.NewMockEvaluationHistoryService(ctrl)
 
 	e := engine.NewExecutor(
 		ctx,
@@ -358,6 +370,8 @@ default allow = true`,
 		providerManager,
 		[]message.HandlerMiddleware{},
 		execMetrics,
+		historyService,
+		&flags.FakeClient{},
 	)
 	require.NoError(t, err, "expected no error")
 

internal/engine/metrics.go

@@ -85,27 +85,27 @@ func (e *ExecutorMetrics) CountEvalStatus(
 // CountRemediationStatus counts remediation events by status.
 func (e *ExecutorMetrics) CountRemediationStatus(
 	ctx context.Context,
-	status string,
+	status db.RemediationStatusTypes,
 	evalID uuid.UUID,
 	projectID uuid.UUID,
 ) {
 	e.evalCounter.Add(ctx, 1, metric.WithAttributes(
 		attribute.String("profile_id", evalID.String()),
 		attribute.String("project_id", projectID.String()),
-		attribute.String("status", status),
+		attribute.String("status", string(status)),
 	))
 }
 
 // CountAlertStatus counts alert events by status.
 func (e *ExecutorMetrics) CountAlertStatus(
 	ctx context.Context,
-	status string,
+	status db.AlertStatusTypes,
 	evalID uuid.UUID,
 	projectID uuid.UUID,
 ) {
 	e.evalCounter.Add(ctx, 1, metric.WithAttributes(
 		attribute.String("profile_id", evalID.String()),
 		attribute.String("project_id", projectID.String()),
-		attribute.String("status", status),
+		attribute.String("status", string(status)),
 	))
 }

internal/flags/constants.go

@@ -18,4 +18,6 @@ package flags
 const (
 	// UserManagement enables user management, i.e. invitations, role assignments, etc.
 	UserManagement Experiment = "user_management"
+	// EvalHistory enables logging of evaluation history in the new tables.
+	EvalHistory Experiment = "eval_history"
 )