monitor

Documentation

Index

• type CheckpointState
• type Client
  • func NewArweaveClient() (Client, error)
  • func NewEthereumClient(endpoint config.Endpoint) (Client, error)
  • func NewFarcasterClient() (Client, error)
  • func NewRssClient(endpoint string, param *config.Parameters) (Client, error)
• 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) 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

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 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 NewRssClient

func NewRssClient(endpoint string, param *config.Parameters) (Client, error)

NewRssClient returns a new rss client.

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) (*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) 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"`
}

func ConstructWorkerProgress

func ConstructWorkerProgress(currentWorkerState, targetWorkerState, latestWorkerState uint64) WorkerProgress

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