runtime

Documentation

Overview

Package runtime replaces the change routines of lib/state and differs in three ways that are the point of the rewrite:

• Reload and Remove affect one environment, where the legacy runtime restarted every ticker of every world on every edit.
• One mutex per environment, so two environments run in parallel.
• State is kept in memory and flushed on an interval, not written whole on every state.set().

The javascript surface is deliberately unchanged, see jsapi.go.

Index

• Constants
• Variables
• func EventTimeProvider(msg platform_connector_lib.EventMsg) (platform_connector_lib.EventMsg, time.Time)
• type BackfillChannelStatus
• type BackfillRangeError
  • func (this *BackfillRangeError) Error() string
• type BackfillState
• type BackfillStatus
• type Runtime
  • func New(config config.Config, environments repo.Environments, states repo.States, ...) *Runtime
  • func (this *Runtime) BackfillStatusOf(id string) (BackfillStatus, error)
  • func (this *Runtime) ExternalDeviceRefs() []string
  • func (this *Runtime) HandleCommand(externalDeviceRef string, externalServiceRef string, cmdMsg interface{}, ...) bool
  • func (this *Runtime) Reload(id string)
  • func (this *Runtime) Remove(id string)
  • func (this *Runtime) SetState(id string, change repo.StateChange) error
  • func (this *Runtime) Snapshot(id string) (StateSnapshot, error)
  • func (this *Runtime) Start(ctx context.Context) error
  • func (this *Runtime) StartBackfill(id string, from time.Time, to time.Time) (BackfillStatus, error)
  • func (this *Runtime) Stop()
• type StateSnapshot

Constants

const EventTimeKey = "moses/event-time-unix-nano"

EventTimeKey is the entry moses puts into an EventMsg to say when the reading was taken. The connector's EventTimeProvider reads it and removes it again, so it never reaches a protocol segment.

An EventMsg is keyed by protocol segment name, so this key shares a namespace with them. The slash makes a collision implausible, and lib.New refuses to start with a protocol segment of this name rather than leaving the collision to be discovered in production.

const MaxBackfillSpan = 366 * 24 * time.Hour

MaxBackfillSpan bounds one window. A year and a day, so that "the last twelve months" is expressible without the caller having to reason about leap years.

Variables

var (
	// ErrBackfillRunning is returned when a job for this environment is still
	// running. Two jobs over overlapping windows would write two readings per
	// instant, and timescale keeps both.
	ErrBackfillRunning = errors.New("a backfill of this environment is already running")

	// ErrNoBackfill is returned when nothing is known about a backfill of this
	// environment. The registry is in memory, so this is also the honest answer
	// after a restart: the job may have completed, may have been interrupted
	// halfway, and this instance cannot tell which.
	ErrNoBackfill = errors.New("nothing is known about a backfill of this environment")
)

var ErrScriptTimeout = errors.New("script exceeded the js timeout")

Functions

func EventTimeProvider

func EventTimeProvider(msg platform_connector_lib.EventMsg) (platform_connector_lib.EventMsg, time.Time)

EventTimeProvider is what lib.New hands the connector. It answers the one question the connector asks before it produces a record: which timestamp the kafka record carries.

This is NOT what stamps the row in timescale. The timescale ingestion never sees this value; it reads the time out of the payload at the service's senergy/time_path, which is why a backfilled reading has to carry its time in the message body as well. See docs/backfill.md.

Types

type BackfillChannelStatus

type BackfillChannelStatus struct {
	ChannelId string `json:"channel_id"`
	AssetId   string `json:"asset_id"`
	Name      string `json:"name"`

	// Backfillable is false when SkipReason says why not.
	Backfillable bool   `json:"backfillable"`
	SkipReason   string `json:"skip_reason,omitempty"`

	// Published counts the readings that reached the platform, Silent the steps
	// that sent nothing - a tick that produced no value at all (a dataset
	// outside its own time range), or one whose value did not move far enough
	// for a channel publishing on change - and Failed the ones the platform
	// refused. The three add up to the steps of the channel's grid.
	Published int64 `json:"published"`
	Silent    int64 `json:"silent,omitempty"`
	Failed    int64 `json:"failed,omitempty"`

	// LastError is the most recent publish failure of this channel, kept so a
	// job that mostly worked still says what went wrong.
	LastError string `json:"last_error,omitempty"`
}

BackfillChannelStatus is what became of one channel of the environment. A channel that was not backfilled says why, because "no data appeared" is otherwise indistinguishable from a channel that published nothing.

type BackfillRangeError

type BackfillRangeError struct {
	Reason string
}

BackfillRangeError is a window that cannot be served, with the reason. The api turns it into a 400.

func (*BackfillRangeError) Error

func (this *BackfillRangeError) Error() string

type BackfillState

type BackfillState string

BackfillState is where a job stands.

const (
	BackfillRunning   BackfillState = "running"
	BackfillDone      BackfillState = "done"
	BackfillFailed    BackfillState = "failed"
	BackfillCancelled BackfillState = "cancelled"
)

type BackfillStatus

type BackfillStatus struct {
	EnvironmentId string        `json:"environment_id"`
	State         BackfillState `json:"state"`

	From time.Time `json:"from"`
	To   time.Time `json:"to"`

	StartedAt  time.Time  `json:"started_at"`
	FinishedAt *time.Time `json:"finished_at,omitempty"`

	// ChannelsTotal counts every channel of the environment, backfillable or
	// not; ChannelsDone counts the ones that are finished with.
	ChannelsTotal int `json:"channels_total"`
	ChannelsDone  int `json:"channels_done"`

	// CurrentChannel and Position are where the job stands right now.
	CurrentChannel string     `json:"current_channel,omitempty"`
	Position       *time.Time `json:"position,omitempty"`

	Published int64  `json:"published"`
	Error     string `json:"error,omitempty"`

	Channels []BackfillChannelStatus `json:"channels"`
}

BackfillStatus is the whole job. It is a copy: the reader never holds a reference into a job that keeps running.

type Runtime

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

Runtime runs every environment of the store.

func New

func New(config config.Config, environments repo.Environments, states repo.States, datasets repo.Datasets, connector *platform_connector_lib.Connector, stateLogger deviceStateLogger) *Runtime

func (*Runtime) BackfillStatusOf

func (this *Runtime) BackfillStatusOf(id string) (BackfillStatus, error)

BackfillStatusOf returns what is known about the backfill of one environment.

func (*Runtime) ExternalDeviceRefs

func (this *Runtime) ExternalDeviceRefs() []string

ExternalDeviceRefs returns every platform device the runtime acts on, sorted. It exists so that the wiring can check the one thing the per world cutover cannot see: a legacy world that was not migrated but whose devices an environment claims anyway.

func (*Runtime) HandleCommand

func (this *Runtime) HandleCommand(externalDeviceRef string, externalServiceRef string, cmdMsg interface{}, responder func(respMsg interface{})) bool

HandleCommand executes the channel a command addresses and reports whether this runtime is responsible for the device at all.

The bool is the cutover: false means the device belongs to no environment, so the caller has to offer the command to the legacy runtime. A device that does belong to an environment is never handed on, even when no channel of it matches the service - otherwise a half migrated document would let both runtimes answer the same command.

func (*Runtime) Reload

func (this *Runtime) Reload(id string)

Reload picks up the current definition of one environment and restarts its channels. Nothing else is touched: the other environments keep ticking, and this environment keeps the runtime state it has in memory, which is newer than what the store holds between two flushes.

func (*Runtime) Remove

func (this *Runtime) Remove(id string)

Remove stops one environment. It is called after its definition was deleted.

func (*Runtime) SetState

func (this *Runtime) SetState(id string, change repo.StateChange) error

SetState merges values into the live state of one running environment. This is how a boundary condition is turned from outside the simulation: an outdoor temperature in the context, a hall temperature on a zone, a machine's speed on an asset. The scripts read it on their next tick.

The change is applied to the in memory state and marked dirty, not written through to the store: the flusher owns that write, and a direct one would be overwritten by it anyway.

func (*Runtime) Snapshot

func (this *Runtime) Snapshot(id string) (StateSnapshot, error)

Snapshot reads the live state of one running environment.

It reports repo.ErrNotRunning for an id this runtime does not hold. That is deliberately not an empty snapshot: an environment that is running and has written nothing yet, and one that is not running at all, are the same three empty maps, and a reader has to be able to tell them apart.

The values are copies. A caller mutating what it got must not reach the maps the scripts keep writing into - which is the same reason the flusher copies before it hands the state to the store.

func (*Runtime) Start

func (this *Runtime) Start(ctx context.Context) error

Start loads every environment and starts its channels. ctx bounds the whole runtime: cancelling it stops every ticker and the flusher, but only Stop writes the state that is still in memory.

func (*Runtime) StartBackfill

func (this *Runtime) StartBackfill(id string, from time.Time, to time.Time) (BackfillStatus, error)

StartBackfill validates a window and starts a job for it.

The validation is synchronous and works on the definition alone, so a caller learns about an impossible window at once. Which channels can actually take a historical timestamp is decided inside the job, because it needs the device types and that is a network read per asset.

func (*Runtime) Stop

func (this *Runtime) Stop()

Stop stops every ticker, ends the flusher and writes what is still dirty.

It is safe to call after ctx of Start has been cancelled, and that is the normal case: the final flush therefore does not use that context.

type StateSnapshot

type StateSnapshot struct {
	State repo.StateChange

	// AsOf is the instant the values were read at, and the instant the values
	// with a time constant were resolved to. It is part of the answer rather
	// than something the reader stamps on arrival: a value that is on its way to
	// a set point means nothing without the moment it was read.
	AsOf time.Time
}

StateSnapshot is the live state of one environment at one instant.

The state is a repo.StateChange and not a repo.RuntimeState on purpose: it is the shape SetState accepts, so a caller reads a value, changes it and sends the same shape back. What RuntimeState carries beyond it - the replay anchors and the approaches in flight - is bookkeeping of the runtime and would only invite a caller to write it back.