indexer

Documentation

Overview

Package indexer provides tools to define, use, and update state indexers.

State service

The state service stores gateway state as keyed blobs. Examples

• IMSI -> directory record blob
• HWID -> gateway status blob

Since state values are stored as arbitrary serialized blobs, the state service has no semantic understanding of stored values. This means searching over stored values would otherwise require an O(n) operation. Examples

• Find IMSI with given IP -- must load all directory records into memory
• Find all gateways that haven't checked in recently -- must load all gateway statuses into memory

Derived state

The solution is to provide customizable, online mechanisms for generating derived state based on existing state. Existing, "primary" state is stored in the state service, and derived, "secondary" state is stored in whichever service owns the derived state. Examples

• Reverse map of directory records
• Primary state: IMSI -> directory record
• Secondary state: IP -> IMSI (stored in e.g. directoryd)
• Reverse map of gateway checkin time
• Primary state: HWID -> gateway status
• Secondary state: checkin time -> HWID (stored in e.g. metricsd)
• List all gateways with multiple kernel versions installed
• Primary state: HWID -> gateway status
• Secondary state: list of gateways (stored in e.g. bootstrapper)

State indexers

State indexers are Orchestrator services registering an IndexerServer under their gRPC endpoint. Any Orchestrator service can provide its own indexer servicer.

The state service discovers indexers using K8s labels. Any service with the label "orc8r.io/state_indexer" will be assumed to provide an indexer servicer.

Indexers provide two additional pieces of metadata -- version and types.

• version: positive integer indicating when the indexer requires reindexing
• types: list of state types the indexer subscribes to

These metadata are indicated by K8s annotations

• orc8r.io/state_indexer_version -- positive integer
• orc8r.io/state_indexer_types -- comma-separated list of state types

Reindexing

When an indexer's implementation changes, its derived state needs to be refreshed. This is accomplished by sending all existing state (of desired types) through the now-updated indexer.

An indexer indicates it needs to undergo a reindex by incrementing its version (exposed via the above-mentioned annotation). From there, the state service automatically handles the reindexing process.

Metrics and logging are available to track long-running reindex processes, as well an indexers CLI which reports desired and current indexer versions.

Implementing a custom indexer

To create a custom indexer, attach an IndexerServer to a new or existing Orchestrator service.

A service can only attach a single indexer. However, that indexer can choose to multiplex its functionality over any desired number of "logical" indexers.

See the orchestrator service for an example custom indexer.

Notes

The state indexer pattern currently provides no mechanism for connecting primary and secondary state. This means secondary state can go stale. Where relevant, consumers of secondary state should take this into account, generally by checking the primary state to ensure it agrees with the secondary state. Examples

• Reverse map of directory records
• Get IMSI from IP -> IMSI map (secondary state)
• Ensure the directory record in the IMSI -> directory map contains the desired IP (primary state)
• Reverse map of gateway checkin time
• Get HWIDs from checkin time -> HWID map (secondary state)
• For each HWID, ensure the gateway status in the HWID -> gateway status map contains the relevant checkin time (primary state)

Automatic reindexing is only supported with Postgres. Deployments targeting Maria will need to use the indexer CLI to manually trigger reindex operations.

There is a trivial but existent race condition during the reindex process. Since the index and reindex operations but use the Index gRPC method, and the index and reindex operations operate in parallel, it's possible for an indexer to receive an outdated piece of state from the reindexer. However, this requires

• reindexer read old state
• new state reported, indexer read new state
• indexer Index call completed
• reindexer Index call completed

If this race condition is intolerable to the desired use case, the solution is to separate out the Index call into Index and Reindex methods. This is not currently implemented as we don't have a concrete use-case for it yet.

Index

• func DeregisterAllForTest(t *testing.T)
• func FilterIDs(types []string, ids []state_types.ID) []state_types.ID
• func FilterStates(types []string, states state_types.StatesByID) state_types.StatesByID
• func MakeProtoInfo(v *Versions) *protos.IndexerInfo
• func MakeProtoInfos(vs []*Versions) map[string]*protos.IndexerInfo
• type Indexer
  • func GetIndexer(serviceName string) (Indexer, error)
  • func GetIndexers() ([]Indexer, error)
  • func GetIndexersForState(stateType string) ([]Indexer, error)
  • func NewRemoteIndexer(serviceName string, version Version, types ...string) Indexer
• type Version
  • func NewIndexerVersion(version int64) (Version, error)
• type Versions
  • func MakeVersion(p *protos.IndexerInfo) *Versions
  • func MakeVersions(ps map[string]*protos.IndexerInfo) []*Versions
  • func (v *Versions) String() string

Functions

func DeregisterAllForTest

func DeregisterAllForTest(t *testing.T)

DeregisterAllForTest deregisters all previously-registered indexers. This should only be called by test code.

func FilterIDs

func FilterIDs(types []string, ids []state_types.ID) []state_types.ID

FilterIDs to the subset that match one of the state types.

func FilterStates

func FilterStates(types []string, states state_types.StatesByID) state_types.StatesByID

FilterStates to the subset that match one of the state types.

func MakeProtoInfo

func MakeProtoInfo(v *Versions) *protos.IndexerInfo

func MakeProtoInfos

func MakeProtoInfos(vs []*Versions) map[string]*protos.IndexerInfo

Types

type Indexer

type Indexer interface {
	// GetID returns the unique identifier for the indexer.
	// For remote indexers, unique ID should be the service name.
	GetID() string

	// GetVersion returns the current version for the indexer.
	// Incrementing the version in a release will result in a reindex.
	// An indexer's version is required to be
	//	- nonzero
	// 	- non-decreasing across successive releases
	GetVersion() Version

	// GetTypes defines the types of states this indexer is interested in.
	GetTypes() []string

	// PrepareReindex prepares for a reindex operation.
	// isFirstReindex is set if this is the first time this indexer has been registered.
	PrepareReindex(from, to Version, isFirstReindex bool) error

	// CompleteReindex indicates the reindex operation is complete.
	CompleteReindex(from, to Version) error

	// Index updates secondary indices based on the added/updated states.
	Index(networkID string, states state_types.StatesByID) (state_types.StateErrors, error)
}

Indexer creates a set of secondary indices for consumption by a service. Each Indexer should

• be owned by a single service
• have its generated data exposed by the owning service, i.e. only one other service should access the generated data directly via the storage interface.

Notes

• There is an unlikely but existent race condition during a reindex operation, where Index could be called with an outdated version of a state. If indexers care about preventing this race condition:
• add a Reindex method to indexer interface, called in-place of Index during reindex operations
• individual indexers should track received state IDs per version and drop Reindex-ed states with stale versions.

func GetIndexer

func GetIndexer(serviceName string) (Indexer, error)

GetIndexer returns the remote indexer for a desired service. Returns nil if not found.

func GetIndexers

func GetIndexers() ([]Indexer, error)

GetIndexers returns all registered indexers.

func GetIndexersForState

func GetIndexersForState(stateType string) ([]Indexer, error)

GetIndexersForState returns all registered indexers which handle the passed state type.

func NewRemoteIndexer

func NewRemoteIndexer(serviceName string, version Version, types ...string) Indexer

NewRemoteIndexer returns an indexer that forwards its methods to the remote indexer servicer.

type Version

type Version uint32

Version of the indexer. Capped to uint32 to fit into Postgres/Maria integer (int32).

func NewIndexerVersion

func NewIndexerVersion(version int64) (Version, error)

NewIndexerVersion returns a new indexer version, ensuring it fits into the required integer size.

type Versions

type Versions struct {
	// IndexerID is the ID of the indexer.
	// ID should be the owning service's name.
	IndexerID string
	// Actual version of the indexer.
	Actual Version
	// Desired version of the indexer.
	Desired Version
}

Versions represents the discrepancy between an indexer's versions, actual vs. desired.

func MakeVersion

func MakeVersion(p *protos.IndexerInfo) *Versions

func MakeVersions

func MakeVersions(ps map[string]*protos.IndexerInfo) []*Versions

func (*Versions) String

func (v *Versions) String() string