upstream

Documentation

Index

• Constants
• func NewPushQueueConsumer(config UpstreamConfig) postq.AsyncEventConsumer
• func NewUpstreamReconciler(upstreamConf UpstreamConfig, pageSize int) *upstreamReconciler
• func PushToUpstream(ctx duty.DBContext, config UpstreamConfig, events []postq.Event) []postq.Event
• func SyncCheckStatuses(ctx context.Context, config UpstreamConfig, batchSize int) error
• type GroupedPushEvents
  • func GroupChangelogsByTables(events []postq.Event) []GroupedPushEvents
• type PaginateRequest
• type PaginateResponse
  • func GetPrimaryKeysHash(ctx context.Context, req PaginateRequest, agentID uuid.UUID) (*PaginateResponse, error)
• type PushData
  • func GetMissingResourceIDs(ctx context.Context, ids []string, paginateReq PaginateRequest) (*PushData, error)
  • func (t *PushData) ApplyLabels(labels map[string]string)
  • func (p *PushData) Attributes() map[string]any
  • func (t *PushData) Count() int
  • func (t *PushData) PopulateAgentID(id uuid.UUID)
  • func (t *PushData) ReplaceTopologyID(id *uuid.UUID)
  • func (p *PushData) String() string
• type UpstreamClient
  • func NewUpstreamClient(config UpstreamConfig) *UpstreamClient
  • func (t *UpstreamClient) Push(ctx context.Context, msg *PushData) error
• type UpstreamConfig
  • func (t *UpstreamConfig) IsPartiallyFilled() bool
  • func (t *UpstreamConfig) LabelsMap() map[string]string
  • func (t *UpstreamConfig) Valid() bool

Constants

const EventPushQueueCreate = "push_queue.create"

Functions

func NewPushQueueConsumer

func NewPushQueueConsumer(config UpstreamConfig) postq.AsyncEventConsumer

NewPushQueueConsumer creates a new push queue consumer

func NewUpstreamReconciler

func NewUpstreamReconciler(upstreamConf UpstreamConfig, pageSize int) *upstreamReconciler

func PushToUpstream

func PushToUpstream(ctx duty.DBContext, config UpstreamConfig, events []postq.Event) []postq.Event

PushToUpstream fetches records specified in events from this instance and sends them to the upstream instance.

func SyncCheckStatuses

func SyncCheckStatuses(ctx context.Context, config UpstreamConfig, batchSize int) error

SyncCheckStatuses pushes check statuses, that haven't already been pushed, to upstream.

Types

type GroupedPushEvents

type GroupedPushEvents struct {
	TableName string
	ItemIDs   [][]string
	Events    postq.Events
}

func GroupChangelogsByTables

func GroupChangelogsByTables(events []postq.Event) []GroupedPushEvents

GroupChangelogsByTables groups the given events by the table they belong to.

type PaginateRequest

type PaginateRequest struct {
	Table string `query:"table"`
	From  string `query:"from"`
	Size  int    `query:"size"`
}

type PaginateResponse

type PaginateResponse struct {
	Hash  string `gorm:"column:sha256sum"`
	Next  string `gorm:"column:last_id"`
	Total int    `gorm:"column:total"`
}

func GetPrimaryKeysHash

func GetPrimaryKeysHash(ctx context.Context, req PaginateRequest, agentID uuid.UUID) (*PaginateResponse, error)

type PushData

type PushData struct {
	AgentName                    string                               `json:"agent_name,omitempty"`
	Canaries                     []models.Canary                      `json:"canaries,omitempty"`
	Checks                       []models.Check                       `json:"checks,omitempty"`
	Components                   []models.Component                   `json:"components,omitempty"`
	ConfigScrapers               []models.ConfigScraper               `json:"config_scrapers,omitempty"`
	ConfigAnalysis               []models.ConfigAnalysis              `json:"config_analysis,omitempty"`
	ConfigChanges                []models.ConfigChange                `json:"config_changes,omitempty"`
	ConfigItems                  []models.ConfigItem                  `json:"config_items,omitempty"`
	CheckStatuses                []models.CheckStatus                 `json:"check_statuses,omitempty"`
	ConfigRelationships          []models.ConfigRelationship          `json:"config_relationships,omitempty"`
	ComponentRelationships       []models.ComponentRelationship       `json:"component_relationships,omitempty"`
	ConfigComponentRelationships []models.ConfigComponentRelationship `json:"config_component_relationships,omitempty"`
	Topologies                   []models.Topology                    `json:"topologies,omitempty"`
}

PushData consists of data about changes to components, configs, analysis.

func GetMissingResourceIDs

func GetMissingResourceIDs(ctx context.Context, ids []string, paginateReq PaginateRequest) (*PushData, error)

func (*PushData) ApplyLabels

func (t *PushData) ApplyLabels(labels map[string]string)

ApplyLabels injects additional labels to the suitable fields

func (*PushData) Attributes

func (p *PushData) Attributes() map[string]any

func (*PushData) Count

func (t *PushData) Count() int

func (*PushData) PopulateAgentID

func (t *PushData) PopulateAgentID(id uuid.UUID)

PopulateAgentID sets agent_id on all the data

func (*PushData) ReplaceTopologyID

func (t *PushData) ReplaceTopologyID(id *uuid.UUID)

ReplaceTopologyID replaces the topology_id for all the components with the provided id.

func (*PushData) String

func (p *PushData) String() string

type UpstreamClient

type UpstreamClient struct {
	*http.Client
}

func NewUpstreamClient

func NewUpstreamClient(config UpstreamConfig) *UpstreamClient

func (*UpstreamClient) Push

func (t *UpstreamClient) Push(ctx context.Context, msg *PushData) error

Push uploads the given push message to the upstream server.

type UpstreamConfig

type UpstreamConfig struct {
	AgentName          string
	Host               string
	InsecureSkipVerify bool
	Username           string
	Password           string
	Labels             []string
}

func (*UpstreamConfig) IsPartiallyFilled

func (t *UpstreamConfig) IsPartiallyFilled() bool

func (*UpstreamConfig) LabelsMap

func (t *UpstreamConfig) LabelsMap() map[string]string

func (*UpstreamConfig) Valid

func (t *UpstreamConfig) Valid() bool