topology

package
v1.0.152 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 27, 2023 License: Apache-2.0 Imports: 32 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultCheckRetentionDays  = 7
	DefaultCanaryRetentionDays = 7
)
View Source 
const DefaultDepth = 3

Variables

View Source 
var (
	CheckRetentionDays  int
	CanaryRetentionDays int
)
View Source 
var CleanupCanaries = &job.Job{
	Name:       "CleanupCanaries",
	Schedule:   "@every 12h",
	Singleton:  true,
	JobHistory: true,
	Retention:  job.Retention3Day,
	RunNow:     true,
	Fn: func(ctx job.JobRuntime) error {
		if CheckRetentionDays <= 0 {
			CheckRetentionDays = DefaultCheckRetentionDays
		}
		tx := ctx.DB().Exec(`
		DELETE FROM canaries
		WHERE
				id NOT IN (SELECT canary_id FROM checks) AND
				(NOW() - deleted_at) > INTERVAL '1 day' * ?
		`, CanaryRetentionDays)

		ctx.History.SuccessCount = int(tx.RowsAffected)
		return tx.Error
	},
}
View Source 
var CleanupChecks = &job.Job{
	Name:       "CleanupChecks",
	Schedule:   "@every 12h",
	Singleton:  true,
	JobHistory: true,
	Retention:  job.Retention3Day,
	Fn: func(ctx job.JobRuntime) error {
		if CheckRetentionDays <= 0 {
			CheckRetentionDays = DefaultCheckRetentionDays
		}
		tx := ctx.DB().Exec(`
					DELETE FROM checks
					WHERE
							id NOT IN (SELECT check_id FROM evidences WHERE check_id IS NOT NULL) AND
							(NOW() - deleted_at) > INTERVAL '1 day' * ?
					`, CheckRetentionDays)

		ctx.History.SuccessCount = int(tx.RowsAffected)
		return tx.Error
	},
}
View Source 
var CleanupMetricsGauges = &job.Job{
	Name:       "CleanupMetricsGauges",
	Schedule:   "@every 1h",
	Singleton:  true,
	JobHistory: true,
	Retention:  job.RetentionDay,
	RunNow:     true,
	Fn: func(ctx job.JobRuntime) error {

		sevenDaysAgo := time.Now().Add(-time.Hour * 24 * 7)
		var deletedCheckIDs []string
		if err := ctx.DB().Model(&models.Check{}).Where("deleted_at > ?", sevenDaysAgo).Pluck("id", &deletedCheckIDs).Error; err != nil {
			return fmt.Errorf("error finding deleted checks: %v", err)
		}

		if ctx.IsDebug() {
			ctx.Debugf("Found %d deleted checks since %s", len(deletedCheckIDs), sevenDaysAgo.Format("2006-01-02 15:04:05"))
		}
		for _, id := range deletedCheckIDs {
			if metrics.Gauge.DeletePartialMatch(prometheus.Labels{"key": id}) > 0 {
				logger.Debugf("Deleted gauge for check: %s", id)
				ctx.History.IncrSuccess()
			}
		}
		return nil
	},
}

CleanupMetricsGauges removes gauges for checks that no longer exist.

View Source 
var ComponentCheckRun = &job.Job{
	Name:       "ComponentCheckRun",
	Schedule:   "@every 2m",
	Singleton:  true,
	JobHistory: true,
	Retention:  job.RetentionHour,
	Fn: func(run job.JobRuntime) error {
		var components = []pkg.Component{}
		if err := run.DB().Table("components").
			Where("component_checks != 'null'").
			Where(duty.LocalFilter).
			Find(&components).Error; err != nil {
			return fmt.Errorf("error getting components: %v", err)
		}

		for _, component := range components {
			relationships, err := GetChecksForComponent(run.Context, &component)
			if err != nil {
				return err
			}
			err = syncCheckComponentRelationships(run.Context, component, relationships)
			if err != nil {
				run.History.AddError(fmt.Sprintf("error persisting relationships: %v", err))
				continue
			}
			run.History.IncrSuccess()
		}
		return nil
	},
}
View Source 
var ComponentConfigRun = &job.Job{
	Name:       "ComponentConfigRun",
	Schedule:   "@every 2m",
	Singleton:  true,
	JobHistory: true,
	Retention:  job.RetentionHour,
	Fn: func(run job.JobRuntime) error {
		db := run.DB().Session(&gorm.Session{NewDB: true})
		var components = []pkg.Component{}
		if err := db.Where(duty.LocalFilter).
			Where("configs != 'null'").
			Select("id", "configs").
			Find(&components).Error; err != nil {
			return fmt.Errorf("error getting components: %v", err)
		}

		for _, component := range components {
			if err := SyncComponentConfigRelationship(db, component); err != nil {
				run.History.AddError(fmt.Sprintf("error persisting config relationships: %v", err))
				continue
			}
			run.History.IncrSuccess()
		}
		return nil
	},
}
View Source 
var ComponentCostRun = &job.Job{
	Name:       "ComponentCostSync",
	JobHistory: true,
	Singleton:  true,
	Retention:  job.RetentionDay,
	Schedule:   "@every 1h",
	Fn: func(ctx job.JobRuntime) error {
		return ctx.DB().Exec(`
				WITH
				component_children AS (
						SELECT components.id, ARRAY(
								SELECT child_id FROM lookup_component_children(components.id::text, -1)
								UNION
								SELECT relationship_id as child_id FROM component_relationships WHERE component_id IN (
										SELECT child_id FROM lookup_component_children(components.id::text, -1)
								)
						) AS child_ids
						FROM components
						GROUP BY components.id
				),
				component_configs AS (
						SELECT component_children.id, ARRAY_AGG(ccr.config_id) as config_ids
						FROM component_children
						INNER JOIN config_component_relationships ccr ON ccr.component_id = ANY(component_children.child_ids)
						GROUP BY component_children.id
				),
				component_config_costs AS (
						SELECT
								component_configs.id,
								SUM(cost_per_minute) AS cost_per_minute,
								SUM(cost_total_1d) AS cost_total_1d,
								SUM(cost_total_7d) AS cost_total_7d,
								SUM(cost_total_30d) AS cost_total_30d
						FROM config_items
						INNER JOIN component_configs ON config_items.id = ANY(component_configs.config_ids)
						GROUP BY component_configs.id
				)

				UPDATE components
				SET
						cost_per_minute = component_config_costs.cost_per_minute,
						cost_total_1d = component_config_costs.cost_total_1d,
						cost_total_7d = component_config_costs.cost_total_7d,
						cost_total_30d = component_config_costs.cost_total_30d
				FROM component_config_costs
				WHERE components.id = component_config_costs.id
				`).Error
	},
}
View Source 
var ComponentRelationshipSync = &job.Job{
	Name:       "ComponentRelationshipSync",
	Schedule:   "@every 5m",
	JobHistory: true,
	Retention:  job.RetentionHour,
	Singleton:  true,
	Fn: func(ctx job.JobRuntime) error {
		var components []models.Component
		if err := ctx.DB().Where(duty.LocalFilter).
			Where("selectors != 'null'").
			Find(&components).Error; err != nil {
			return fmt.Errorf("error getting components: %v", err)
		}

		for _, component := range components {
			hash := component.Selectors.Hash()
			comps, err := duty.FindComponents(ctx.Context, component.Selectors, duty.PickColumns("id", "path"))
			if err != nil {
				ctx.History.AddError(fmt.Sprintf("error getting components with selectors: %s. err: %v", component.Selectors, err))
				continue
			}
			relationships := []models.ComponentRelationship{}
			for _, c := range comps {
				relationships = append(relationships, models.ComponentRelationship{
					RelationshipID:   component.ID,
					ComponentID:      c.ID,
					SelectorID:       hash,
					RelationshipPath: component.Path + "." + component.ID.String(),
				})
			}

			err = syncComponentRelationships(ctx.Context, component.ID, relationships)
			if err != nil {
				ctx.History.AddError(fmt.Sprintf("error syncing relationships: %v", err))
				continue
			}
			ctx.History.IncrSuccess()
		}
		return nil
	},
}
View Source 
var ComponentStatusSummarySync = &job.Job{
	Name:       "ComponentStatusSummarySync",
	Schedule:   "@every 2m",
	JobHistory: true,
	Retention:  job.RetentionHour,
	Singleton:  true,
	Fn: func(ctx job.JobRuntime) error {
		topology, err := Query(ctx.Context, duty.TopologyOptions{Depth: 3})
		if err != nil {
			return fmt.Errorf("error getting components: %v", err)
		}

		for _, c := range topology.Components {
			tx := ctx.DB().Where("id = ? and (status != ? or summary != ?)", c.ID, c.Status, c.Summary).
				UpdateColumns(models.Component{Status: c.Status, Summary: c.Summary})
			if tx.Error != nil {
				ctx.History.AddError(tx.Error.Error())
			} else {
				ctx.History.IncrSuccess()
			}
		}

		return nil
	},
}
View Source 
var Jobs = []*job.Job{
	ComponentConfigRun,
	ComponentCheckRun,
	CleanupCanaries,
	CleanupChecks,
	CleanupMetricsGauges,
	ComponentCostRun,
	ComponentRelationshipSync,
	ComponentStatusSummarySync,
}
View Source 
var SyncComponentRelationships2 = &job.Job{
	Name: "ComponentStatusSummarySync",
	Fn: func(ctx job.JobRuntime) error {
		return nil
	},
}

Functions

func GetChecksForComponent added in v1.0.152 

func GetChecksForComponent(ctx context.Context, component *pkg.Component) ([]models.CheckComponentRelationship, error)

func NewTopologyParams 

func NewTopologyParams(values url.Values) duty.TopologyOptions

func PersistConfigComponentRelationship added in v1.0.152 

func PersistConfigComponentRelationship(db *gorm.DB, configID, componentID uuid.UUID, selectorID string) error

func Query 

func Query(ctx dutyContext.Context, params duty.TopologyOptions) (*duty.TopologyResponse, error)

func Run 

func Run(opts TopologyRunOptions, t v1.Topology) ([]*pkg.Component, models.JobHistory)

func SyncComponentConfigRelationship added in v1.0.152 

func SyncComponentConfigRelationship(db *gorm.DB, component pkg.Component) error

func SyncComponents added in v0.38.257 

func SyncComponents(opts TopologyRunOptions, topology v1.Topology) (int, error)

Types

type ComponentContext added in v0.38.135 

type ComponentContext struct {
	*context.KubernetesContext
	Topology     v1.Topology
	ComponentAPI v1.Component
	// Components keep track of the components that properties can apply to,
	// properties can return a map of component names to properties to facilitate
	// queries that are more efficient to perform for all components rather than a component at a time
	Components *pkg.Components
	// Properties can either be looked up on an individual component, or act as a summary across all components
	CurrentComponent *pkg.Component

	JobHistory *models.JobHistory
	Duty       dutyContext.Context
	DB         *gorm.DB
	// contains filtered or unexported fields
}

func NewComponentContext added in v0.38.135 

func NewComponentContext(ctx dutyContext.Context, system v1.Topology) *ComponentContext

func (*ComponentContext) Clone added in v0.38.135 

func (c *ComponentContext) Clone() *ComponentContext

func (*ComponentContext) GetTemplater added in v0.38.193 

func (c *ComponentContext) GetTemplater() ktemplate.StructTemplater

func (*ComponentContext) SetCurrentComponent added in v0.38.193 

func (c *ComponentContext) SetCurrentComponent(component *pkg.Component)

func (*ComponentContext) String added in v1.0.152 

func (c *ComponentContext) String() string

func (*ComponentContext) TemplateComponent added in v0.38.193 

func (c *ComponentContext) TemplateComponent(component *v1.ComponentSpec) error

func (*ComponentContext) TemplateConfig added in v0.38.193 

func (c *ComponentContext) TemplateConfig(config *types.ConfigQuery) error

func (*ComponentContext) TemplateProperty added in v0.38.193 

func (c *ComponentContext) TemplateProperty(property *v1.Property) error

func (*ComponentContext) TemplateStruct added in v0.38.193 

func (c *ComponentContext) TemplateStruct(data interface{}) error

func (*ComponentContext) WithComponents added in v0.38.135 

func (c *ComponentContext) WithComponents(components *pkg.Components, current *pkg.Component) *ComponentContext

type TopologyRunOptions 

type TopologyRunOptions struct {
	dutyContext.Context
	Depth     int
	Namespace string
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.