Merge pull request #18 from erlandf/add-steps-support

feat: add steps from pipelineactivity

collector/pipeline_activity.go

@@ -48,7 +48,20 @@ func (c *PipelineActivityCollector) Start(ctx context.Context) error { // nolint
 
 	return nil
 }
+func SimplifyStep(coreStep jenkinsv1.CoreActivityStep) store.SimplifiedActivityStep {
 
+	if coreStep.Status == "" || coreStep.StartedTimestamp == nil || coreStep.CompletedTimestamp == nil {
+		return store.SimplifiedActivityStep{}
+	}
+
+	return store.SimplifiedActivityStep{
+		Name:               coreStep.Name,
+		Status:             coreStep.Status.String(),
+		StartedTimestamp:   coreStep.StartedTimestamp.Time,
+		CompletedTimestamp: coreStep.CompletedTimestamp.Time,
+		Duration:           coreStep.CompletedTimestamp.Time.Sub(coreStep.StartedTimestamp.Time),
+	}
+}
 func (c *PipelineActivityCollector) storePipeline(pa *jenkinsv1.PipelineActivity) {
 	if pa == nil {
 		return
@@ -79,6 +92,38 @@ func (c *PipelineActivityCollector) storePipeline(pa *jenkinsv1.PipelineActivity
 		return
 	}
 
+	var simplifiedSteps []store.SimplifiedActivityStep
+	for _, step := range pa.Spec.Steps {
+		log.WithField("step", step.Kind).Trace("Simplifying step")
+		if step.Kind == "Stage" {
+			for _, stageStep := range step.Stage.Steps {
+				simplifiedStep := SimplifyStep(stageStep)
+				if simplifiedStep.Name == "" {
+					log.WithField("step", step.Kind).Trace("Ignoring empty step")
+				} else {
+					simplifiedSteps = append(simplifiedSteps, simplifiedStep)
+				}
+			}
+			simplifiedSteps = append(simplifiedSteps, SimplifyStep(step.Stage.CoreActivityStep))
+		}
+		if step.Kind == "Promote" {
+			simplifiedStep := SimplifyStep(step.Promote.CoreActivityStep)
+			if simplifiedStep.Name == "" {
+				log.WithField("step", step.Kind).Trace("Ignoring empty step")
+			} else {
+				simplifiedSteps = append(simplifiedSteps, simplifiedStep)
+			}
+		}
+		if step.Kind == "Preview" {
+			simplifiedStep := SimplifyStep(step.Preview.CoreActivityStep)
+			if simplifiedStep.Name == "" {
+				log.WithField("step", step.Kind).Trace("Ignoring empty step")
+			} else {
+				simplifiedSteps = append(simplifiedSteps, simplifiedStep)
+			}
+		}
+	}
+	log.WithField("steps", len(simplifiedSteps)).Trace("Simplified steps")
 	pipeline := store.Pipeline{
 		Owner:      pa.Spec.GitOwner,
 		Repository: pa.Spec.GitRepository,
@@ -87,6 +132,7 @@ func (c *PipelineActivityCollector) storePipeline(pa *jenkinsv1.PipelineActivity
 		Author:     pa.Spec.Author,
 		StartTime:  pa.Spec.StartedTimestamp.Time.In(time.UTC),
 		EndTime:    pa.Spec.CompletedTimestamp.Time.In(time.UTC),
+		Steps:      simplifiedSteps,
 	}
 	pipeline.Duration = pipeline.EndTime.Sub(pipeline.StartTime)
 

store/pipeline.go

@@ -12,6 +12,14 @@ import (
 
 type PipelineType string
 
+type SimplifiedActivityStep struct {
+	Name               string
+	Status             string
+	StartedTimestamp   time.Time
+	CompletedTimestamp time.Time
+	Duration           time.Duration
+}
+
 const (
 	PipelineTypeRelease     = PipelineType("release")
 	PipelineTypePullRequest = PipelineType("pullrequest")
@@ -29,6 +37,7 @@ type Pipeline struct {
 	StartTime   time.Time
 	EndTime     time.Time
 	Duration    time.Duration
+	Steps       []SimplifiedActivityStep
 }
 
 type PipelineStore struct {
@@ -56,6 +65,20 @@ func (s *PipelineStore) Migrations() []migration.Func {
 				duration bigint NOT NULL,
 				CONSTRAINT pipeline_pkey PRIMARY KEY (type, owner, repository, pull_request, context, build)
 			);
+			CREATE TABLE pipelinesteps (
+				type VARCHAR NOT NULL,
+				owner VARCHAR NOT NULL,
+				repository VARCHAR NOT NULL,
+				pull_request int,
+				context VARCHAR NOT NULL,
+				build int NOT NULL,
+				step_name VARCHAR NOT NULL,
+				step_status VARCHAR NOT NULL,
+				step_started_time timestamp without time zone NOT NULL,
+				step_completed_time timestamp without time zone NOT NULL,
+				step_duration bigint NOT NULL,
+				CONSTRAINT pipelinesteps_pkey PRIMARY KEY (type, owner, repository, pull_request, context, build, step_name)
+			);
 		`),
 	}
 }
@@ -74,6 +97,12 @@ func (s *PipelineStore) Add(ctx context.Context, p Pipeline) error {
 		return fmt.Errorf("failed to add pipeline: %w", err)
 	}
 
+	for _, step := range p.Steps {
+		_, err = tx.Exec(ctx, "INSERT INTO pipelinesteps (type, owner, repository, pull_request, context, build, step_name, step_status, step_started_time, step_completed_time, step_duration) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING;", p.Type, p.Owner, p.Repository, p.PullRequest, p.Context, p.Build, step.Name, step.Status, step.StartedTimestamp, step.CompletedTimestamp, step.Duration.Seconds())
+		if err != nil {
+			return fmt.Errorf("failed to add pipeline step: %w", err)
+		}
+	}
 	if err = tx.Commit(ctx); err != nil {
 		return fmt.Errorf("failed to commit insertion of pipeline: %w", err)
 	}