Documentation
¶
Index ¶
- Constants
- Variables
- func NotifyArtifactDeleted(ctx context.Context, broker Broker, assetID uuid.UUID)
- func NotifyAssetTicketSettingsChanged(ctx context.Context, broker Broker, assetID uuid.UUID)
- func NotifyAssetVersionDeleted(ctx context.Context, broker Broker, assetID uuid.UUID)
- func NotifyDependencyVulnsDetected(ctx context.Context, broker Broker, artifact ArtifactPayload, ...)
- func NotifyFirstPartyScanned(ctx context.Context, broker Broker, assetVersion ArtifactPayload, ...)
- func NotifySBOMCreatedWithVulns(ctx context.Context, broker Broker, artifact ArtifactPayload, ...)
- func NotifyVEXRuleCreated(ctx context.Context, broker Broker, assetID uuid.UUID)
- func NotifyVEXRuleDeleted(ctx context.Context, broker Broker, assetID uuid.UUID)
- func NotifyVulnEventCreated(ctx context.Context, broker Broker, assetID uuid.UUID, ...)
- type ArtifactPayload
- type AssetPayload
- type AssetPipelinePayload
- type Broker
- type EmptyPayload
- type EventBrokerImpl
- type EventBrokerStub
- func (EventBrokerStub) FinishedSince(ctx context.Context, topic NotifyTopic, since time.Time, match any) ([]json.RawMessage, error)
- func (EventBrokerStub) IsHealthy() bool
- func (EventBrokerStub) Listen(topic NotifyTopic, handler NotifyHandler) error
- func (EventBrokerStub) Notify(ctx context.Context, tx pgx.Tx, topic NotifyTopic, payload any, ...) error
- func (EventBrokerStub) Publish(ctx context.Context, topic PubSubTopic, payload any) error
- func (EventBrokerStub) Schedule(topic NotifyTopic, cronExpr string) error
- func (EventBrokerStub) Subscribe(topic PubSubTopic, handler func(ctx context.Context, payload json.RawMessage)) error
- type InlineBroker
- type ListeningConnection
- type NotifyHandler
- type NotifyOptions
- type NotifyTopic
- type Payload
- type PostgreSQLBroker
- func (b *PostgreSQLBroker) GetActiveTopics() []PubSubTopic
- func (b *PostgreSQLBroker) IsHealthy() bool
- func (b *PostgreSQLBroker) Publish(ctx context.Context, topic PubSubTopic, payload any) error
- func (b *PostgreSQLBroker) SetShouldReceiveOwnMessages(should bool)
- func (b *PostgreSQLBroker) Subscribe(topic PubSubTopic, handler SubscribeHandler) error
- type PostgreSQLMessage
- type PubSubTopic
- type Queue
- func (r *Queue) FinishedSince(ctx context.Context, topic NotifyTopic, since time.Time, match any) ([]json.RawMessage, error)
- func (r *Queue) Listen(topic NotifyTopic, handler NotifyHandler) error
- func (r *Queue) Notify(ctx context.Context, tx pgx.Tx, topic NotifyTopic, payload any, ...) error
- func (r *Queue) Schedule(topic NotifyTopic, cronExpr string) error
- type SubscribeHandler
Constants ¶
const HandlerTimeout = 2 * time.Hour
const QueueAssetPipeline = "asset_pipeline"
QueueAssetPipeline runs the per asset pipeline steps, separated so they cannot starve the other daemons
Variables ¶
var BrokerStub = EventBrokerStub{}
Functions ¶
func NotifyArtifactDeleted ¶
NotifyArtifactDeleted - an artifact of the asset was deleted
func NotifyAssetTicketSettingsChanged ¶
NotifyAssetTicketSettingsChanged - the ticket thresholds or the ticket range of the asset changed
func NotifyAssetVersionDeleted ¶
NotifyAssetVersionDeleted - asset versions of the asset were deleted
func NotifyDependencyVulnsDetected ¶
func NotifyDependencyVulnsDetected(ctx context.Context, broker Broker, artifact ArtifactPayload, vulnIDs []uuid.UUID)
NotifyDependencyVulnsDetected - a scan of the artifact opened the vulns
func NotifyFirstPartyScanned ¶
func NotifyFirstPartyScanned(ctx context.Context, broker Broker, assetVersion ArtifactPayload, openedVulnIDs []uuid.UUID)
NotifyFirstPartyScanned - first party scan results of the asset version were stored, the scan opened the vulns
func NotifySBOMCreatedWithVulns ¶
func NotifySBOMCreatedWithVulns(ctx context.Context, broker Broker, artifact ArtifactPayload, vulnIDs []uuid.UUID)
NotifySBOMCreatedWithVulns - an sbom of the artifact was created or changed and scanned, the scan opened the vulns
func NotifyVEXRuleCreated ¶
NotifyVEXRuleCreated - a vex rule of the asset was created and applied
func NotifyVEXRuleDeleted ¶
NotifyVEXRuleDeleted - a vex rule of the asset was deleted and the vulns it covered were reopened
func NotifyVulnEventCreated ¶
func NotifyVulnEventCreated(ctx context.Context, broker Broker, assetID uuid.UUID, eventType dtos.VulnEventType)
NotifyVulnEventCreated - a vuln of the asset received an event. If the event changes the risk, the statistics are recalculated right away
Types ¶
type ArtifactPayload ¶
type ArtifactPayload struct {
AssetID uuid.UUID `json:"assetId"`
AssetVersionName string `json:"assetVersionName"`
ArtifactName string `json:"artifactName,omitempty"`
UserAgent string `json:"userAgent,omitempty"`
// the vulns a scan opened
VulnIDs []uuid.UUID `json:"vulnIds,omitempty"`
}
ArtifactPayload names the artifact an event happened on - the listeners load everything else themselves, so a retry never works on stale data. ArtifactName is empty for events of the whole asset version.
func (ArtifactPayload) AlertContext ¶
func (p ArtifactPayload) AlertContext() monitoring.AlertContext
type AssetPayload ¶
AssetPayload names the asset an event happened on
func (AssetPayload) AlertContext ¶
func (p AssetPayload) AlertContext() monitoring.AlertContext
type AssetPipelinePayload ¶
AssetPipelinePayload is the payload of every asset pipeline step job. The asset is the only argument - jobs are unique by their arguments, therefore an asset has at most one queued or running job per step. Every step enqueues the following step once it succeeded.
func (AssetPipelinePayload) AlertContext ¶
func (p AssetPipelinePayload) AlertContext() monitoring.AlertContext
type Broker ¶
type Broker interface {
// pub/sub - broadcast to all instances of devguard
Publish(ctx context.Context, topic PubSubTopic, payload any) error
Subscribe(topic PubSubTopic, handler SubscribeHandler) error
// deliver exactly once - only a single instance of devguard will receive the message and execute the handler.
Notify(ctx context.Context, tx pgx.Tx, topic NotifyTopic, payload any, opts *NotifyOptions) error
Listen(topic NotifyTopic, handler NotifyHandler) error
// Schedule notifies the topic according to the cron expression - exactly once across all instances.
Schedule(topic NotifyTopic, cronExpr string) error
// FinishedSince returns the distinct payloads of the jobs of the topic which completed or were discarded after since.
// match filters by payload containment (jsonb @>) - nil matches every job.
// finished jobs are only kept for completedJobRetention, since must not lie further back.
FinishedSince(ctx context.Context, topic NotifyTopic, since time.Time, match any) ([]json.RawMessage, error)
IsHealthy() bool
}
combines pub/sub and notify/listen
type EmptyPayload ¶
type EmptyPayload struct{}
EmptyPayload is the payload of the jobs which need none, e.g. the scheduled daemons
func (EmptyPayload) AlertContext ¶
func (EmptyPayload) AlertContext() monitoring.AlertContext
type EventBrokerImpl ¶
type EventBrokerImpl struct {
*PostgreSQLBroker
*Queue
}
type EventBrokerStub ¶
type EventBrokerStub struct {
}
func (EventBrokerStub) FinishedSince ¶
func (EventBrokerStub) FinishedSince(ctx context.Context, topic NotifyTopic, since time.Time, match any) ([]json.RawMessage, error)
func (EventBrokerStub) IsHealthy ¶
func (EventBrokerStub) IsHealthy() bool
func (EventBrokerStub) Listen ¶
func (EventBrokerStub) Listen(topic NotifyTopic, handler NotifyHandler) error
func (EventBrokerStub) Notify ¶
func (EventBrokerStub) Notify(ctx context.Context, tx pgx.Tx, topic NotifyTopic, payload any, opts *NotifyOptions) error
func (EventBrokerStub) Publish ¶
func (EventBrokerStub) Publish(ctx context.Context, topic PubSubTopic, payload any) error
func (EventBrokerStub) Schedule ¶
func (EventBrokerStub) Schedule(topic NotifyTopic, cronExpr string) error
func (EventBrokerStub) Subscribe ¶
func (EventBrokerStub) Subscribe(topic PubSubTopic, handler func(ctx context.Context, payload json.RawMessage)) error
type InlineBroker ¶
type InlineBroker struct {
EventBrokerStub
// contains filtered or unexported fields
}
InlineBroker runs the handlers of a notification right away in the caller - tests use it to keep queued work synchronous. Like a job, a failing handler does not fail the notification.
func NewInlineBroker ¶
func NewInlineBroker() *InlineBroker
func (*InlineBroker) Listen ¶
func (b *InlineBroker) Listen(topic NotifyTopic, handler NotifyHandler) error
func (*InlineBroker) Notify ¶
func (b *InlineBroker) Notify(ctx context.Context, tx pgx.Tx, topic NotifyTopic, payload any, opts *NotifyOptions) error
type ListeningConnection ¶
type ListeningConnection struct {
Conn *pgxpool.Conn
Subscribers []SubscribeHandler
}
type NotifyHandler ¶
type NotifyHandler = func(ctx context.Context, payload json.RawMessage) error
type NotifyOptions ¶
type NotifyOptions = river.InsertOpts
type NotifyTopic ¶
type NotifyTopic string
const ( TopicSBOMCreated NotifyTopic = "artifact.sbomCreated" TopicDependencyVulnsDetected NotifyTopic = "artifact.dependencyVulnsDetected" TopicArtifactDeleted NotifyTopic = "artifact.deleted" TopicFirstPartyScanned NotifyTopic = "assetVersion.firstPartyScanned" TopicAssetVersionDeleted NotifyTopic = "assetVersion.deleted" TopicVEXRuleCreated NotifyTopic = "vexRule.created" TopicVEXRuleDeleted NotifyTopic = "vexRule.deleted" TopicAssetTicketSettingsChanged NotifyTopic = "asset.ticketSettingsChanged" )
Domain events state what happened - the listeners decide what follows, see daemons.RegisterJobs. They are notified after the change is committed: a listener might run before an open transaction finishes. Notifying is best effort - a failure is logged, the sweep of the asset pipeline catches up on the follow-up work.
const ( TopicVEXRuleRecommendations NotifyTopic = "daemon.vexRuleRecommendations" TopicCleanupOrphanedRecords NotifyTopic = "daemon.cleanupOrphanedRecords" TopicSBOMGarbageCollection NotifyTopic = "daemon.sbomGarbageCollection" TopicOpenSourceInsights NotifyTopic = "daemon.openSourceInsights" TopicExternalEntityGarbage NotifyTopic = "daemon.externalEntityGarbage" TopicAssetPipeline NotifyTopic = "daemon.assetPipeline" TopicVulnDB NotifyTopic = "daemon.vulnDB" TopicScan NotifyTopic = "daemon.scan" TopicFixedVersions NotifyTopic = "daemon.fixedVersions" TopicDirectDependencyFixedVersion NotifyTopic = "daemon.directDependencyFixedVersion" // asset pipeline steps - one job per asset and step, the payload is an AssetPipelinePayload. // they run on QueueAssetPipeline, see assetPipelineTopicPrefix TopicPipelineDeleteOldAssetVersions NotifyTopic = "pipeline.deleteOldAssetVersions" TopicPipelineSyncUpstream NotifyTopic = "pipeline.syncUpstream" TopicPipelineResolveLicenses NotifyTopic = "pipeline.resolveLicenses" TopicPipelineFindLicenseRisks NotifyTopic = "pipeline.findLicenseRisks" TopicPipelineApplyVEXRules NotifyTopic = "pipeline.applyVEXRules" TopicPipelineRecalculateRisk NotifyTopic = "pipeline.recalculateRisk" TopicPipelineResolveFixedVersions NotifyTopic = "pipeline.resolveFixedVersions" TopicPipelineSyncTickets NotifyTopic = "pipeline.syncTickets" TopicPipelineResolveTicketDifferences NotifyTopic = "pipeline.resolveTicketDifferences" TopicPipelineCollectStats NotifyTopic = "pipeline.collectStats" )
Job topics. They are persisted as river job kinds - do not rename while jobs of the old name may still be queued.
type Payload ¶
type Payload interface {
AlertContext() monitoring.AlertContext
}
Payload is implemented by every job payload - a failing job names the context of its payload in the alert
type PostgreSQLBroker ¶
type PostgreSQLBroker struct {
ID string // Unique identifier for the broker instance
// contains filtered or unexported fields
}
PostgreSQLBroker implements the Broker interface using PostgreSQL LISTEN/NOTIFY
func NewPostgreSQLBroker ¶
func NewPostgreSQLBroker(db *pgxpool.Pool) (*PostgreSQLBroker, error)
NewPostgreSQLBroker creates a new PostgreSQL broker
func (*PostgreSQLBroker) GetActiveTopics ¶
func (b *PostgreSQLBroker) GetActiveTopics() []PubSubTopic
GetActiveTopics returns a list of topics currently being listened to
func (*PostgreSQLBroker) IsHealthy ¶
func (b *PostgreSQLBroker) IsHealthy() bool
IsHealthy checks if the broker is functioning properly
func (*PostgreSQLBroker) Publish ¶
func (b *PostgreSQLBroker) Publish(ctx context.Context, topic PubSubTopic, payload any) error
Publish implements the Broker interface
func (*PostgreSQLBroker) SetShouldReceiveOwnMessages ¶
func (b *PostgreSQLBroker) SetShouldReceiveOwnMessages(should bool)
func (*PostgreSQLBroker) Subscribe ¶
func (b *PostgreSQLBroker) Subscribe(topic PubSubTopic, handler SubscribeHandler) error
type PostgreSQLMessage ¶
type PostgreSQLMessage struct {
ID string `json:"id"`
Topic PubSubTopic `json:"topic"`
Payload json.RawMessage `json:"payload"`
Timestamp time.Time `json:"timestamp"`
SenderID string `json:"sender_id,omitempty"` // Optional field for sender ID
}
type PubSubTopic ¶
type PubSubTopic string
const ( PolicyChange PubSubTopic = "policyChange" VulnDBUpdated PubSubTopic = "vulnDBUpdated" )
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
func (*Queue) FinishedSince ¶
func (r *Queue) FinishedSince(ctx context.Context, topic NotifyTopic, since time.Time, match any) ([]json.RawMessage, error)
func (*Queue) Listen ¶
func (r *Queue) Listen(topic NotifyTopic, handler NotifyHandler) error
It must be called before the fx app is started - river does not support adding workers to a running client.
func (*Queue) Notify ¶
func (r *Queue) Notify(ctx context.Context, tx pgx.Tx, topic NotifyTopic, payload any, opts *NotifyOptions) error
func (*Queue) Schedule ¶
func (r *Queue) Schedule(topic NotifyTopic, cronExpr string) error
Schedule enqueues a job for the topic according to the cron expression (standard 5 fields, e.g. "0 * * * *"). Only the river leader schedules jobs - therefore it runs once per schedule across all instances. On startup or leader change the job is enqueued right away - unless it already ran within the current period (e.g. the current hour for an hourly schedule). This way a new instance does not wait for its first run, but deploys do not trigger a run either. A new job is not enqueued while the previous one is still pending, running or retrying.
type SubscribeHandler ¶
type SubscribeHandler = func(ctx context.Context, payload json.RawMessage)