-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1995 from opengovern/feat-query-validator
fix: add query validator scheduled
- Loading branch information
Showing
19 changed files
with
1,163 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "os" | ||
| "os/signal" | ||
| "syscall" | ||
|
|
||
| queryvalidator "github.com/opengovern/opengovernance/pkg/query-validator" | ||
| ) | ||
|
|
||
| func main() { | ||
| ctx := context.Background() | ||
| ctx, cancel := context.WithCancel(ctx) | ||
|
|
||
| c := make(chan os.Signal, 1) | ||
| signal.Notify(c, os.Interrupt, syscall.SIGTERM) | ||
| defer func() { | ||
| signal.Stop(c) | ||
| cancel() | ||
| }() | ||
|
|
||
| go func() { | ||
| select { | ||
| case <-c: | ||
| cancel() | ||
| case <-ctx.Done(): | ||
| } | ||
| }() | ||
|
|
||
| if err := queryvalidator.WorkerCommand().ExecuteContext(ctx); err != nil { | ||
| fmt.Println(err) | ||
| os.Exit(1) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| ARG PLUGIN_REGISTRY | ||
| FROM ${PLUGIN_REGISTRY}/steampipe-plugin-aws:0.0.1 as aws | ||
| FROM ${PLUGIN_REGISTRY}/steampipe-plugin-azure:0.0.1 as azure | ||
| FROM ${PLUGIN_REGISTRY}/steampipe-plugin-azuread:0.0.1 as azuread | ||
| FROM ${PLUGIN_REGISTRY}/steampipe-plugin-opengovernance:0.0.1 as opengovernance | ||
|
|
||
| FROM ubuntu:20.04 AS base | ||
| RUN apt-get update && apt-get install -y \ | ||
| curl \ | ||
| sudo \ | ||
| && rm -rf /var/lib/apt/lists/* | ||
|
|
||
| RUN sudo /bin/sh -c "$(curl -fsSL https://steampipe.io/install/steampipe.sh)" | ||
|
|
||
| COPY --from=aws /steampipe-plugin-aws.plugin /home/steampipe/.steampipe/plugins/hub.steampipe.io/plugins/turbot/aws@latest/steampipe-plugin-aws.plugin | ||
| COPY --from=azure /steampipe-plugin-azure.plugin /home/steampipe/.steampipe/plugins/hub.steampipe.io/plugins/turbot/azure@latest/steampipe-plugin-azure.plugin | ||
| COPY --from=azuread /steampipe-plugin-azuread.plugin /home/steampipe/.steampipe/plugins/hub.steampipe.io/plugins/turbot/azuread@latest/steampipe-plugin-azuread.plugin | ||
| COPY --from=opengovernance /steampipe-plugin-opengovernance.plugin /home/steampipe/.steampipe/plugins/local/opengovernance/opengovernance.plugin | ||
|
|
||
| USER root | ||
| RUN useradd -ms /bin/bash steampipe | ||
| RUN mkdir -p /home/steampipe/.steampipe/config | ||
| RUN mkdir -p /home/steampipe/.steampipe/db | ||
| RUN mkdir -p /home/steampipe/.steampipe/db/14.2.0 | ||
| RUN chown -R steampipe:steampipe /home/steampipe | ||
| RUN chmod -R 755 /home/steampipe | ||
| RUN apt update | ||
| RUN apt install -y procps htop | ||
| USER steampipe | ||
|
|
||
| RUN steampipe plugin list | ||
|
|
||
| COPY ./build/query-validator-worker / | ||
|
|
||
| ENTRYPOINT [ "/query-validator-worker" ] | ||
| CMD [ "/query-validator-worker" ] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| package model | ||
|
|
||
| import ( | ||
| queryvalidator "github.com/opengovern/opengovernance/pkg/query-validator" | ||
| "gorm.io/gorm" | ||
| ) | ||
|
|
||
| type QueryValidatorJob struct { | ||
| gorm.Model | ||
| QueryId string | ||
| QueryType queryvalidator.QueryType | ||
| Status queryvalidator.QueryValidatorStatus | ||
| FailureMessage string | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| package db | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
| queryrunner "github.com/opengovern/opengovernance/services/inventory/query-runner" | ||
|
|
||
| "github.com/opengovern/opengovernance/pkg/describe/db/model" | ||
| queryvalidator "github.com/opengovern/opengovernance/pkg/query-validator" | ||
| "gorm.io/gorm" | ||
| ) | ||
|
|
||
| func (db Database) CreateQueryValidatorJob(job *model.QueryValidatorJob) (uint, error) { | ||
| tx := db.ORM.Create(job) | ||
| if tx.Error != nil { | ||
| return 0, tx.Error | ||
| } | ||
|
|
||
| return job.ID, nil | ||
| } | ||
|
|
||
| func (db Database) GetQueryValidatorJob(id uint) (*model.QueryValidatorJob, error) { | ||
| var job model.QueryValidatorJob | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Where("id = ?", id).First(&job) | ||
| if tx.Error != nil { | ||
| return nil, tx.Error | ||
| } | ||
| return &job, nil | ||
| } | ||
|
|
||
| func (db Database) ListQueryValidatorJobsById(ids []string) ([]model.QueryValidatorJob, error) { | ||
| var jobs []model.QueryValidatorJob | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Where("id IN ?", ids).Find(&jobs) | ||
| if tx.Error != nil { | ||
| return nil, tx.Error | ||
| } | ||
| return jobs, nil | ||
| } | ||
|
|
||
| func (db Database) ListQueryValidatorJobs() ([]model.QueryValidatorJob, error) { | ||
| var jobs []model.QueryValidatorJob | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Find(&jobs) | ||
| if tx.Error != nil { | ||
| return nil, tx.Error | ||
| } | ||
| return jobs, nil | ||
| } | ||
|
|
||
| func (db Database) FetchCreatedQueryValidatorJobs(limit int64) ([]model.QueryValidatorJob, error) { | ||
| var jobs []model.QueryValidatorJob | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Where("status = ?", queryvalidator.QueryValidatorCreated).Limit(int(limit)).Find(&jobs) | ||
| if tx.Error != nil { | ||
| return nil, tx.Error | ||
| } | ||
| return jobs, nil | ||
| } | ||
|
|
||
| func (db Database) GetInProgressJobsCount() (int64, error) { | ||
| var count int64 | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Where("status IN ?", []string{string(queryvalidator.QueryValidatorInProgress), | ||
| string(queryvalidator.QueryValidatorQueued)}).Count(&count) | ||
| if tx.Error != nil { | ||
| return 0, tx.Error | ||
| } | ||
| return count, nil | ||
| } | ||
|
|
||
| func (db Database) DeleteQueryValidatorJob(id uint) error { | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Delete(&model.QueryValidatorJob{}, id) | ||
| if tx.Error != nil { | ||
| return tx.Error | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (db Database) UpdateQueryValidatorJobStatus(jobId uint, status queryvalidator.QueryValidatorStatus, failureReason string) error { | ||
| tx := db.ORM.Model(&model.QueryValidatorJob{}).Where("id = ?", jobId). | ||
| Updates(model.QueryValidatorJob{Status: status, FailureMessage: failureReason}) | ||
| if tx.Error != nil { | ||
| return tx.Error | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (db Database) UpdateTimedOutInProgressQueryValidators() error { | ||
| tx := db.ORM. | ||
| Model(&model.QueryValidatorJob{}). | ||
| Where("status = ?", queryrunner.QueryRunnerInProgress). | ||
| Where("updated_at < NOW() - INTERVAL '5 MINUTES'"). | ||
| Updates(model.QueryValidatorJob{Status: queryvalidator.QueryValidatorTimeOut, FailureMessage: "Job timed out"}) | ||
| if tx.Error != nil { | ||
| return tx.Error | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (db Database) UpdateTimedOutQueuedQueryValidators() error { | ||
| tx := db.ORM. | ||
| Model(&model.QueryValidatorJob{}). | ||
| Where("status = ?", queryrunner.QueryRunnerQueued). | ||
| Where("updated_at < NOW() - INTERVAL '12 HOURS'"). | ||
| Updates(model.QueryValidatorJob{Status: queryvalidator.QueryValidatorTimeOut, FailureMessage: "Job timed out"}) | ||
| if tx.Error != nil { | ||
| return tx.Error | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (db Database) ListQueryValidatorJobForInterval(interval, triggerType, createdBy string) ([]model.QueryValidatorJob, error) { | ||
| var job []model.QueryValidatorJob | ||
|
|
||
| tx := db.ORM.Model(&model.QueryValidatorJob{}) | ||
|
|
||
| if interval != "" { | ||
| tx = tx.Where(fmt.Sprintf("NOW() - updated_at < INTERVAL '%s'", interval)) | ||
| } | ||
| if triggerType != "" { | ||
| tx = tx.Where("trigger_type = ?", triggerType) | ||
| } | ||
| if createdBy != "" { | ||
| tx = tx.Where("created_by = ?", createdBy) | ||
| } | ||
|
|
||
| tx = tx.Find(&job) | ||
|
|
||
| if tx.Error != nil { | ||
| if errors.Is(tx.Error, gorm.ErrRecordNotFound) { | ||
| return nil, nil | ||
| } | ||
| return nil, tx.Error | ||
| } | ||
| return job, nil | ||
| } | ||
|
|
||
| func (db Database) CleanupAllQueryValidatorJobs() error { | ||
| tx := db.ORM.Where("1 = 1").Unscoped().Delete(&model.QueryValidatorJob{}) | ||
| if tx.Error != nil { | ||
| return tx.Error | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| package query_runner | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "github.com/nats-io/nats.go/jetstream" | ||
| queryvalidator "github.com/opengovern/opengovernance/pkg/query-validator" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| func (s *JobScheduler) RunQueryRunnerReportJobResultsConsumer(ctx context.Context) error { | ||
| if _, err := s.jq.Consume(ctx, "scheduler-query-validator", queryvalidator.StreamName, []string{queryvalidator.JobResultQueueTopic}, "scheduler-query-validator", func(msg jetstream.Msg) { | ||
| if err := msg.Ack(); err != nil { | ||
| s.logger.Error("Failed committing message", zap.Error(err)) | ||
| } | ||
|
|
||
| var result queryvalidator.JobResult | ||
| if err := json.Unmarshal(msg.Data(), &result); err != nil { | ||
| s.logger.Error("Failed to unmarshal ComplianceReportJob results", zap.Error(err)) | ||
| return | ||
| } | ||
|
|
||
| s.logger.Info("Processing ReportJobResult for Job", | ||
| zap.Uint("jobId", result.ID), | ||
| zap.String("status", string(result.Status)), | ||
| ) | ||
| err := s.db.UpdateQueryValidatorJobStatus(result.ID, result.Status, result.FailureMessage) | ||
| if err != nil { | ||
| s.logger.Error("Failed to update the status of QueryRunnerReportJob", | ||
| zap.Uint("jobId", result.ID), | ||
| zap.Error(err)) | ||
| return | ||
| } | ||
| }); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| <-ctx.Done() | ||
| return nil | ||
| } |
Oops, something went wrong.