upstream

Documentation

Index

• Constants
• func NewPushQueueConsumer(config UpstreamConfig) postq.AsyncEventConsumer
• func NewUpstreamReconciler(upstreamConf UpstreamConfig, pageSize int) *upstreamReconciler
• func Push(ctx context.Context, config UpstreamConfig, msg *PushData) error
• func PushToUpstream(ctx duty.DBContext, config UpstreamConfig, events []postq.Event) []postq.Event
• type GroupedPushEvents
  • func GroupChangelogsByTables(events []postq.Event) []GroupedPushEvents
• type PaginateRequest
• type PaginateResponse
  • func GetPrimaryKeysHash(ctx duty.DBContext, req PaginateRequest, agentID uuid.UUID) (*PaginateResponse, error)
• type PushData
  • func GetMissingResourceIDs(ctx duty.DBContext, ids []string, paginateReq PaginateRequest) (*PushData, error)
  • func (t *PushData) ApplyLabels(labels map[string]string)
  • 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 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 Push

func Push(ctx context.Context, config UpstreamConfig, msg *PushData) error

Push uploads the given push message to the upstream server.

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.

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 duty.DBContext, 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 duty.DBContext, 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) 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 UpstreamConfig

type UpstreamConfig struct {
	AgentName string
	Host      string
	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