purge

Documentation

Index

• func ConfigureManager(man *cereal.Manager, workflowName cereal.WorkflowName, opts ...TaskOpt) error
• func CreateOrUpdatePurgeWorkflow(ctx context.Context, man *cereal.Manager, scheduleName string, ...) error
• type EsPolicy
  • func (p EsPolicy) Purge(ctx context.Context, esSidecarClient es.EsSidecarServiceClient) error
• type PgPolicy
  • func (p PgPolicy) Purge(ctx context.Context) error
• type Policies
  • func NewPolicies() *Policies
• type Server
  • func NewServer(man *cereal.Manager, scheduleName string, workflowName cereal.WorkflowName, ...) (*Server, error)
  • func (server *Server) Configure(ctx context.Context, req *data_lifecycle.ConfigureRequest) (*data_lifecycle.ConfigureResponse, error)
  • func (server *Server) Run(ctx context.Context, _ *data_lifecycle.RunRequest) (*data_lifecycle.RunResponse, error)
  • func (server *Server) Show(ctx context.Context, req *data_lifecycle.ShowRequest) (*data_lifecycle.ShowResponse, error)
• type ServerOpt
  • func WithServerEsSidecarClient(esClient es.EsSidecarServiceClient) ServerOpt
• type Task
  • func (t *Task) Run(ctx context.Context, task cereal.Task) (interface{}, error)
• type TaskOpt
  • func WithTaskEsSidecarClient(client es.EsSidecarServiceClient) TaskOpt
• type Workflow
  • func (s *Workflow) OnCancel(w cereal.WorkflowInstance, ev cereal.CancelEvent) cereal.Decision
  • func (s *Workflow) OnStart(w cereal.WorkflowInstance, ev cereal.StartEvent) cereal.Decision
  • func (s *Workflow) OnTaskComplete(w cereal.WorkflowInstance, ev cereal.TaskCompleteEvent) cereal.Decision

Functions

func ConfigureManager

func ConfigureManager(man *cereal.Manager, workflowName cereal.WorkflowName, opts ...TaskOpt) error

ConfigureManager registers the purge workflow executor and task executor.

func CreateOrUpdatePurgeWorkflow

func CreateOrUpdatePurgeWorkflow(
	ctx context.Context,
	man *cereal.Manager,
	scheduleName string,
	workflowName cereal.WorkflowName,
	defaultPolicies *Policies,
	enabled bool,
	recurrence *rrule.RRule) error

CreateOrUpdatePurgeWorkflow creates a purge scheduled workflow. If a matching purge workflow exists it update the policies with any new default policies that don't exist. It also updates existing elasticsearch policies index names and purge fields in case they have changed.

Types

type EsPolicy

type EsPolicy struct {
	Name          string `json:"name"`
	IndexName     string `json:"index_name"`
	OlderThanDays int32  `json:"older_than_days"`
	// If the custom purge field is set we'll delete via document instead of
	// timeseries index.
	CustomPurgeField string `json:"custom_purge_field"`
	Disabled         bool   `json:"disabled"`
}

EsPolicy represents an elasticsearch purge policy

func (EsPolicy) Purge

func (p EsPolicy) Purge(ctx context.Context, esSidecarClient es.EsSidecarServiceClient) error

Purge purges based on the policy

type PgPolicy

type PgPolicy struct {
	Name     string `json:"name"`
	Disabled bool   `json:"disabled"`
}

NOTE: PgPolicies are just a shim until they are actually required by a service. Policy represents a postgres purge policy

func (PgPolicy) Purge

func (p PgPolicy) Purge(ctx context.Context) error

Purge is where we'll do the purging

type Policies

type Policies struct {
	Es map[string]EsPolicy `json:"es"`
	Pg map[string]PgPolicy `json:"pg"`
}

Policies represent the purge policies that are persisted in the workflow parameters.

func NewPolicies

func NewPolicies() *Policies

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server is the purge server implementation

func NewServer

func NewServer(
	man *cereal.Manager,
	scheduleName string,
	workflowName cereal.WorkflowName,
	defaultPolicies *Policies,
	opts ...ServerOpt) (*Server, error)

NewServer - create a new purge gRPC server

func (*Server) Configure

func (server *Server) Configure(ctx context.Context,
	req *data_lifecycle.ConfigureRequest) (*data_lifecycle.ConfigureResponse, error)

Configure creates or updates the current policies. Every policy that matches the update policy name will be updated or created if it doesn't exist. If a policy update does not match an allowed policy an error will be returned.

func (*Server) Run

func (server *Server) Run(
	ctx context.Context,
	_ *data_lifecycle.RunRequest) (*data_lifecycle.RunResponse, error)

Run executes the purge workflow

func (*Server) Show

func (server *Server) Show(ctx context.Context,
	req *data_lifecycle.ShowRequest) (*data_lifecycle.ShowResponse, error)

Show shows the current scheduled purge job

type ServerOpt

type ServerOpt func(*Server)

func WithServerEsSidecarClient

func WithServerEsSidecarClient(esClient es.EsSidecarServiceClient) ServerOpt

type Task

type Task struct {
	EsSidecarClient es.EsSidecarServiceClient
}

func (*Task) Run

func (t *Task) Run(ctx context.Context, task cereal.Task) (interface{}, error)

type TaskOpt

type TaskOpt func(*Task)

func WithTaskEsSidecarClient

func WithTaskEsSidecarClient(client es.EsSidecarServiceClient) TaskOpt

type Workflow

type Workflow struct{}

func (*Workflow) OnCancel

func (s *Workflow) OnCancel(w cereal.WorkflowInstance, ev cereal.CancelEvent) cereal.Decision

func (*Workflow) OnStart

func (s *Workflow) OnStart(w cereal.WorkflowInstance, ev cereal.StartEvent) cereal.Decision

func (*Workflow) OnTaskComplete

func (s *Workflow) OnTaskComplete(w cereal.WorkflowInstance, ev cereal.TaskCompleteEvent) cereal.Decision