monitor

Documentation

Index

• Constants
• Variables
• type CheckpointState
• type Client
  • func NewActivityPubClient(network network.Network, param *config.Parameters) (Client, error)
  • func NewArweaveClient() (Client, error)
  • func NewEthereumClient(endpoint config.Endpoint) (Client, error)
  • func NewFarcasterClient() (Client, error)
  • func NewNearClient(endpoint config.Endpoint) (Client, error)
• type CronJob
  • func NewCronJob(client rueidis.Client, name string, timeout time.Duration) (*CronJob, error)
  • func (c *CronJob) AddFunc(ctx context.Context, spec string, cmd func()) error
  • func (c *CronJob) ReleaseLock()
  • func (c *CronJob) Renewal(ctx context.Context)
  • func (c *CronJob) Start()
  • func (c *CronJob) Stop()
• type Monitor
  • func NewMonitor(_ context.Context, configFile *config.File, databaseClient database.Client, ...) (*Monitor, error)
  • func (m *Monitor) GetWorkerStatusByID(ctx context.Context, workerID string) workerx.Status
  • func (m *Monitor) MaintainCoveragePeriod(ctx context.Context) error
  • func (m *Monitor) MonitorMockWorkerStatus(ctx context.Context, currentState CheckpointState, ...) error
  • func (m *Monitor) MonitorWorkerStatus(ctx context.Context) error
  • func (m *Monitor) Run(ctx context.Context) error
  • func (m *Monitor) UpdateWorkerProgress(ctx context.Context, workerID string, progress WorkerProgress) error
  • func (m *Monitor) UpdateWorkerStatusByID(ctx context.Context, workerID string, status string) error
• type WorkerProgress
  • func ConstructWorkerProgress(currentWorkerState, targetWorkerState, latestWorkerState uint64, ...) WorkerProgress

Constants

const (
	MonitorWorkerStatusJob = "worker_status"
	DatabaseMaintenanceJob = "database_maintenance"
)

Variables

var KeyPrefix = "cronjob:%s"

Types

type CheckpointState

type CheckpointState struct {
	BlockHeight      uint64 `json:"block_height"`
	BlockTimestamp   uint64 `json:"block_timestamp"`
	BlockNumber      uint64 `json:"block_number"`
	EventID          uint64 `json:"event_id"`
	CastsBackfill    bool   `json:"casts_backfill"`
	ReactionBackfill bool   `json:"reaction_backfill"`
}

type Client

type Client interface {
	// CurrentState returns the current block number (ethereum), height (arweave) or event id (farcaster) of the client from Checkpoints table in database.
	CurrentState(state CheckpointState) (uint64, uint64)
	// TargetState checks if the target block number/height is set in the parameters.
	TargetState(param *config.Parameters) (uint64, uint64)
	// LatestState returns the latest block number (ethereum), height (arweave) or event id (farcaster) or err (rss) of the client from network rpc/api.
	LatestState(ctx context.Context) (uint64, uint64, error)
}

func NewActivityPubClient

func NewActivityPubClient(network network.Network, param *config.Parameters) (Client, error)

NewActivityPubClient returns a new ActivityPub client.

func NewArweaveClient

func NewArweaveClient() (Client, error)

NewArweaveClient returns a new arweave client.

func NewEthereumClient

func NewEthereumClient(endpoint config.Endpoint) (Client, error)

NewEthereumClient returns a new ethereum client.

func NewFarcasterClient

func NewFarcasterClient() (Client, error)

NewFarcasterClient returns a new farcaster client.

func NewNearClient

func NewNearClient(endpoint config.Endpoint) (Client, error)

type CronJob

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

func NewCronJob

func NewCronJob(client rueidis.Client, name string, timeout time.Duration) (*CronJob, error)

func (*CronJob) AddFunc

func (c *CronJob) AddFunc(ctx context.Context, spec string, cmd func()) error

func (*CronJob) ReleaseLock

func (c *CronJob) ReleaseLock()

func (*CronJob) Renewal

func (c *CronJob) Renewal(ctx context.Context)

func (*CronJob) Start

func (c *CronJob) Start()

func (*CronJob) Stop

func (c *CronJob) Stop()

type Monitor

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

func NewMonitor

func NewMonitor(_ context.Context, configFile *config.File, databaseClient database.Client, redisClient rueidis.Client, networkParamsCaller *vsl.NetworkParamsCaller, settlementCaller *vsl.SettlementCaller) (*Monitor, error)

NewMonitor creates a new monitor instance.

func (*Monitor) GetWorkerStatusByID

func (m *Monitor) GetWorkerStatusByID(ctx context.Context, workerID string) workerx.Status

GetWorkerStatusByID gets worker status from Redis cache by network and workerName.

func (*Monitor) MaintainCoveragePeriod

func (m *Monitor) MaintainCoveragePeriod(ctx context.Context) error

func (*Monitor) MonitorMockWorkerStatus

func (m *Monitor) MonitorMockWorkerStatus(ctx context.Context, currentState CheckpointState, targetWorkerState, latestState uint64) error

func (*Monitor) MonitorWorkerStatus

func (m *Monitor) MonitorWorkerStatus(ctx context.Context) error

MonitorWorkerStatus checks the worker status by comparing the current and latest block height/number. flags the worker as unhealthy if it's left behind the latest block height/number by more than the tolerance.

func (*Monitor) Run

func (m *Monitor) Run(ctx context.Context) error

func (*Monitor) UpdateWorkerProgress

func (m *Monitor) UpdateWorkerProgress(ctx context.Context, workerID string, progress WorkerProgress) error

UpdateWorkerProgress updates worker progress (state) in each monitoring cycle to Redis Cache.

func (*Monitor) UpdateWorkerStatusByID

func (m *Monitor) UpdateWorkerStatusByID(ctx context.Context, workerID string, status string) error

UpdateWorkerStatusByID updates the worker status in the Redis cache by id.

type WorkerProgress

type WorkerProgress struct {
	RemoteState  uint64 `json:"remote_state"`
	IndexedState uint64 `json:"indexed_state"`
	IndexCount   int64  `json:"index_count"`
}

func ConstructWorkerProgress

func ConstructWorkerProgress(currentWorkerState, targetWorkerState, latestWorkerState uint64, indexCount int64) WorkerProgress

ConstructWorkerProgress constructs the worker progress from current, target and latest block height/number.