iotanalytics

package
v1.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 25 Imported by: 0

README

IoT Analytics

Parity grade: A · SDK aws-sdk-go-v2/service/iotanalytics@v1.32.0 · last audited 2026-07-24 (be69d5ece)

Coverage

Metric Value
Operations audited 34 (34 ok)
Feature families 2 (2 ok)
Known gaps 3
Deferred items 0
Resource leaks clean
Known gaps
  • GetDatasetContent always returns an empty entries array (no S3-backed data URIs) since this backend has no S3 delivery integration -- consistent with CreateDatasetContent's synchronous SUCCEEDED simulation, not tracked as a bug.
  • items_still_open: RunPipelineActivity lambda/deviceRegistryEnrich/deviceShadowEnrich activities remain pass-through, not real invocations. Reason: real AWS invokes a Lambda function / looks up AWS IoT Device Registry or Device Shadow data for these, which requires cross-service calls this backend has no wiring for -- iotanalytics's Provider.Init (provider.go) receives only a *service.AppContext (JanitorCtx/Logger/PortAlloc), not a shared backend registry like cloudformation's ResourceCreator (which threads rc.backends.Lambda through at a higher app-assembly layer this task was explicitly barred from touching, i.e. cli.go). Wiring that through would be a cross-cutting architecture change outside a single-service parity pass.
  • items_still_open: RunPipelineActivity filter/math expression language (pipeline_expr.go) implements literals, message-attribute identifiers, arithmetic (+ - * / %), comparison (= != <> < <= > >=), logical (AND/OR/NOT), and parentheses -- covering AWS's documented examples (e.g. "temp > 50", "(temp - 32) / 1.8"). It does NOT implement SQL function calls (TRIM/SUBSTR/date functions/etc.), LIKE, IN, or BETWEEN. Reason: AWS's real filter/math grammar is an undocumented open SQL-like superset with no published formal grammar; the implemented subset is a real, tested evaluator (not a stub) covering the operators AWS's own docs demonstrate, but a full function library is unbounded scope this pass did not attempt rather than risk an incorrect partial implementation of unspecified semantics.

More

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrChannelNotFound is returned when a channel does not exist.
	ErrChannelNotFound = newNotFoundError("channel not found")
	// ErrDatastoreNotFound is returned when a datastore does not exist.
	ErrDatastoreNotFound = newNotFoundError("datastore not found")
	// ErrDatasetNotFound is returned when a dataset does not exist.
	ErrDatasetNotFound = newNotFoundError("dataset not found")
	// ErrPipelineNotFound is returned when a pipeline does not exist.
	ErrPipelineNotFound = newNotFoundError("pipeline not found")
	// ErrDatasetContentNotFound is returned when a dataset content version does not exist.
	ErrDatasetContentNotFound = newNotFoundError("dataset content not found")
	// ErrLoggingOptionsNotFound is returned when logging options have not been configured.
	ErrLoggingOptionsNotFound = newNotFoundError("logging options not found")
	// ErrReprocessingNotFound is returned when a pipeline reprocessing job does not exist.
	ErrReprocessingNotFound = newNotFoundError("reprocessing not found")
	// ErrResourceNotFound is returned when a tagged resource ARN does not match any known resource.
	ErrResourceNotFound = newNotFoundError("resource not found")
	// ErrAlreadyExists is returned when a resource with the given name already exists.
	ErrAlreadyExists = errors.New("resource already exists")
	// ErrValidation is returned when request input fails validation.
	ErrValidation = errors.New("validation error")
)

Sentinel errors for IoT Analytics backend operations.

View Source 
var ErrExprSyntax = errors.New("pipeline expression error")

ErrExprSyntax is returned when a pipeline filter/math expression fails to parse or evaluate.

View Source 
var ErrNilAppContext = errors.New("AppContext is required")

ErrNilAppContext is returned when Provider.Init is called with a nil AppContext.

View Source 
var ErrNoSnapshot = errors.New("backend does not support restore")

ErrNoSnapshot is returned when a backend does not support snapshot/restore.

Functions

This section is empty.

Types

type AttributePartition 

type AttributePartition struct {
	AttributeName string `json:"attributeName"`
}

AttributePartition defines a datastore partition by message attribute.

type BatchPutMessageErrorEntry 

type BatchPutMessageErrorEntry struct {
	ChannelName  string `json:"-"`
	ErrorCode    string `json:"errorCode,omitempty"`
	ErrorMessage string `json:"errorMessage,omitempty"`
	MessageID    string `json:"messageId,omitempty"`
}

BatchPutMessageErrorEntry is a per-message error in BatchPutMessage.

type Channel 

type Channel struct {
	Tags                   map[string]string `json:"tags"`
	Storage                *ChannelStorage   `json:"storage,omitempty"`
	RetentionPeriod        *RetentionPeriod  `json:"retentionPeriod,omitempty"`
	Name                   string            `json:"name"`
	ARN                    string            `json:"arn"`
	Status                 string            `json:"status"`
	CreationTime           float64           `json:"creationTime"`
	LastUpdate             float64           `json:"lastUpdate"`
	LastMessageArrivalTime float64           `json:"lastMessageArrivalTime,omitempty"`
}

Channel stores all metadata and state for a single IoT Analytics channel.

type ChannelMessage 

type ChannelMessage struct {
	MessageID string
	Payload   []byte
}

ChannelMessage stores a single message ingested into a channel.

type ChannelStorage 

type ChannelStorage struct {
	ServiceManagedS3  *ServiceManagedS3Storage         `json:"serviceManagedS3,omitempty"`
	CustomerManagedS3 *CustomerManagedS3ChannelStorage `json:"customerManagedS3,omitempty"`
}

ChannelStorage is the storage configuration for a channel.

type ColumnSchema 

type ColumnSchema struct {
	Name string `json:"name"`
	Type string `json:"type"`
}

ColumnSchema defines a column in a Parquet schema.

type ContentDeliveryDestination 

type ContentDeliveryDestination struct {
	IotEventsDestinationConfiguration *IotEventsDestination       `json:"iotEventsDestinationConfiguration,omitempty"`
	S3DestinationConfiguration        *S3DestinationConfiguration `json:"s3DestinationConfiguration,omitempty"`
}

ContentDeliveryDestination is the destination for a content delivery rule.

type ContentDeliveryRule 

type ContentDeliveryRule struct {
	Destination *ContentDeliveryDestination `json:"destination"`
	EntryName   string                      `json:"entryName,omitempty"`
}

ContentDeliveryRule defines where dataset content is delivered on creation.

type CustomerManagedS3ChannelStorage 

type CustomerManagedS3ChannelStorage struct {
	Bucket    string `json:"bucket"`
	KeyPrefix string `json:"keyPrefix,omitempty"`
	RoleArn   string `json:"roleArn"`
}

CustomerManagedS3ChannelStorage is customer-managed S3 for channels.

type CustomerManagedS3DatastoreStorage 

type CustomerManagedS3DatastoreStorage struct {
	Bucket    string `json:"bucket"`
	KeyPrefix string `json:"keyPrefix,omitempty"`
	RoleArn   string `json:"roleArn"`
}

CustomerManagedS3DatastoreStorage is customer-managed S3 for datastores.

type Dataset 

type Dataset struct {
	Tags                    map[string]string        `json:"tags"`
	VersioningConfiguration *VersioningConfiguration `json:"versioningConfiguration,omitempty"`
	Name                    string                   `json:"name"`
	ARN                     string                   `json:"arn"`
	Status                  string                   `json:"status"`
	Actions                 []DatasetAction          `json:"actions,omitempty"`
	Triggers                []DatasetTrigger         `json:"triggers,omitempty"`
	ContentDeliveryRules    []ContentDeliveryRule    `json:"contentDeliveryRules,omitempty"`
	LateDataRules           []LateDataRule           `json:"lateDataRules,omitempty"`
	CreationTime            float64                  `json:"creationTime"`
	LastUpdate              float64                  `json:"lastUpdate"`
}

Dataset stores all metadata and state for a single IoT Analytics dataset.

type DatasetAction 

type DatasetAction struct {
	QueryAction     *DatasetQueryAction     `json:"queryAction,omitempty"`
	ContainerAction *DatasetContainerAction `json:"containerAction,omitempty"`
	ActionName      string                  `json:"actionName"`
}

DatasetAction is an action on a dataset (query or container).

type DatasetContainerAction 

type DatasetContainerAction struct {
	Image                 string                 `json:"image"`
	ExecutionRoleArn      string                 `json:"executionRoleArn"`
	ResourceConfiguration *ResourceConfiguration `json:"resourceConfiguration"`
	Variables             []DatasetVariable      `json:"variables,omitempty"`
}

DatasetContainerAction defines a container execution action on a dataset.

type DatasetContent 

type DatasetContent struct {
	VersionID      string  `json:"versionId"`
	Status         string  `json:"status"`
	CreationTime   float64 `json:"creationTime"`
	CompletionTime float64 `json:"completionTime"`
	ScheduleTime   float64 `json:"scheduleTime"`
}

DatasetContent stores a single content version of an IoT Analytics dataset.

ScheduleTime is the time the content generation was scheduled to start (AWS docs: "the time the creation of the dataset contents was scheduled to start", distinct from CreationTime, "the actual time the creation ... was started"). This backend only creates dataset content synchronously via a direct CreateDatasetContent call (there is no background cron-trigger simulation for DatasetTrigger.Schedule), so ScheduleTime is always set equal to CreationTime -- the same behavior AWS exhibits for a manually invoked CreateDatasetContent that wasn't fired by a schedule trigger.

type DatasetQueryAction 

type DatasetQueryAction struct {
	SQLQuery string               `json:"sqlQuery"`
	Filters  []DatasetQueryFilter `json:"filters,omitempty"`
}

DatasetQueryAction defines an SQL query action on a dataset.

type DatasetQueryFilter 

type DatasetQueryFilter struct {
	DeltaTime *DeltaTime `json:"deltaTime,omitempty"`
}

DatasetQueryFilter is a filter applied to a query action.

type DatasetTrigger 

type DatasetTrigger struct {
	Schedule *ScheduleExpression    `json:"schedule,omitempty"`
	Dataset  *DatasetTriggerDataset `json:"dataset,omitempty"`
}

DatasetTrigger triggers automatic dataset content creation.

type DatasetTriggerDataset 

type DatasetTriggerDataset struct {
	Name string `json:"name"`
}

DatasetTriggerDataset triggers a dataset when another dataset produces content.

type DatasetVariable 

type DatasetVariable struct {
	StringValue                *string  `json:"stringValue,omitempty"`
	DoubleValue                *float64 `json:"doubleValue,omitempty"`
	DatasetContentVersionValue *string  `json:"datasetContentVersionValue,omitempty"`
	OutputFileURIValue         *string  `json:"outputFileUriValue,omitempty"`
	Name                       string   `json:"name"`
}

DatasetVariable is a variable passed to a container action.

type Datastore 

type Datastore struct {
	Tags                    map[string]string        `json:"tags"`
	Storage                 *DatastoreStorage        `json:"storage,omitempty"`
	RetentionPeriod         *RetentionPeriod         `json:"retentionPeriod,omitempty"`
	FileFormatConfiguration *FileFormatConfiguration `json:"fileFormatConfiguration,omitempty"`
	Partitions              *DatastorePartitions     `json:"partitions,omitempty"`
	Name                    string                   `json:"name"`
	ARN                     string                   `json:"arn"`
	Status                  string                   `json:"status"`
	CreationTime            float64                  `json:"creationTime"`
	LastUpdate              float64                  `json:"lastUpdate"`
	LastMessageArrivalTime  float64                  `json:"lastMessageArrivalTime,omitempty"`
}

Datastore stores all metadata and state for a single IoT Analytics datastore.

type DatastorePartitionEntry 

type DatastorePartitionEntry struct {
	AttributePartition *AttributePartition `json:"attributePartition,omitempty"`
	TimestampPartition *TimestampPartition `json:"timestampPartition,omitempty"`
}

DatastorePartitionEntry is one partition definition (union).

type DatastorePartitions 

type DatastorePartitions struct {
	Partitions []DatastorePartitionEntry `json:"partitions"`
}

DatastorePartitions holds all partition definitions for a datastore.

type DatastoreStorage 

type DatastoreStorage struct {
	ServiceManagedS3             *ServiceManagedS3Storage           `json:"serviceManagedS3,omitempty"`
	CustomerManagedS3            *CustomerManagedS3DatastoreStorage `json:"customerManagedS3,omitempty"`
	IotSiteWiseMultiLayerStorage *IotSiteWiseMultiLayerStorage      `json:"iotSiteWiseMultiLayerStorage,omitempty"`
}

DatastoreStorage is the storage configuration for a datastore.

type DeltaTime 

type DeltaTime struct {
	TimeExpression string `json:"timeExpression"`
	OffsetSeconds  int    `json:"offsetSeconds"`
}

DeltaTime defines an offset for dataset query filters.

type DeltaTimeSessionWindowConfiguration 

type DeltaTimeSessionWindowConfiguration struct {
	TimeoutInMinutes int `json:"timeoutInMinutes"`
}

DeltaTimeSessionWindowConfiguration defines a session window for late data.

type FileFormatConfiguration 

type FileFormatConfiguration struct {
	JSONConfiguration    *JSONConfiguration    `json:"jsonConfiguration,omitempty"`
	ParquetConfiguration *ParquetConfiguration `json:"parquetConfiguration,omitempty"`
}

FileFormatConfiguration defines the file format for a datastore.

type GlueConfiguration 

type GlueConfiguration struct {
	TableName    string `json:"tableName"`
	DatabaseName string `json:"databaseName"`
}

GlueConfiguration defines AWS Glue catalog settings for S3 delivery.

type Handler 

type Handler struct {
	Backend StorageBackend
	// contains filtered or unexported fields
}

Handler is the HTTP handler for the IoT Analytics REST API.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new IoT Analytics handler with a pre-built dispatch table.

func (*Handler) ChaosOperations 

func (h *Handler) ChaosOperations() []string

ChaosOperations returns all operations that can be fault-injected.

func (*Handler) ChaosRegions 

func (h *Handler) ChaosRegions() []string

ChaosRegions returns all regions this handler handles.

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

ChaosServiceName returns the lowercase AWS service name for fault rule matching.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation extracts the IoT Analytics operation name from the request.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource extracts the resource name from the URL path.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations returns the list of supported IoT Analytics operations.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler function for IoT Analytics requests.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all backend state.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence by delegating to the backend if it supports it.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher returns a function that matches IoT Analytics REST API requests.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence by delegating to the backend if it supports it.

type InMemoryBackend 

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

InMemoryBackend is the in-memory backend for IoT Analytics.

channels, datastores, datasets, and pipelines are each a *store.Table[T] (see store_setup.go): all four key off a real, non-json:"-" identity field the value type already carries (Name), so none need a DTO wrapper, and none need a secondary store.Index since nothing in this backend does an ARN-keyed reverse lookup against them (resolveARNResource parses the resource name back out of the ARN string and looks it up directly). tags, channelMessages, and datasetContents are left as plain maps: none of their value types is a *T (map[string]string, [][]byte, and []*DatasetContent respectively), so none fits store.Table's keyed-by-single-identity-value shape. See persistence.go for how they round-trip alongside the registered tables.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new in-memory IoT Analytics backend with a background service context.

func NewInMemoryBackendWithContext 

func NewInMemoryBackendWithContext(svcCtx context.Context) *InMemoryBackend

NewInMemoryBackendWithContext creates a new in-memory IoT Analytics backend whose background goroutines are bounded by svcCtx. If svcCtx is nil, context.Background is used.

func (*InMemoryBackend) AddChannelInternal 

func (b *InMemoryBackend) AddChannelInternal(name string) *Channel

AddChannelInternal seeds a channel by name (test helper).

func (*InMemoryBackend) AddDatasetInternal 

func (b *InMemoryBackend) AddDatasetInternal(name string) *Dataset

AddDatasetInternal seeds a dataset by name (test helper).

func (*InMemoryBackend) AddDatastoreInternal 

func (b *InMemoryBackend) AddDatastoreInternal(name string) *Datastore

AddDatastoreInternal seeds a datastore by name (test helper).

func (*InMemoryBackend) AddPipelineInternal 

func (b *InMemoryBackend) AddPipelineInternal(name string) *Pipeline

AddPipelineInternal seeds a pipeline by name (test helper).

func (*InMemoryBackend) BatchPutMessage 

func (b *InMemoryBackend) BatchPutMessage(
	channelName string,
	messages []messageInput,
) ([]BatchPutMessageErrorEntry, error)

BatchPutMessage ingests messages into a channel. Validates count ≤ 100, per-message payload ≤ 128 KB, messageId ≤ 128 chars, and total batch ≤ 500 KB.

func (*InMemoryBackend) CancelPipelineReprocessing 

func (b *InMemoryBackend) CancelPipelineReprocessing(pipelineName, reprocessingID string) error

CancelPipelineReprocessing cancels a running pipeline reprocessing job.

func (*InMemoryBackend) CreateChannel 

func (b *InMemoryBackend) CreateChannel(
	ctx context.Context,
	name string,
	tags map[string]string,
	storage *ChannelStorage,
	retention *RetentionPeriod,
) (*Channel, error)

CreateChannel creates a new IoT Analytics channel.

func (*InMemoryBackend) CreateDataset 

func (b *InMemoryBackend) CreateDataset(
	ctx context.Context,
	name string,
	tags map[string]string,
	actions []DatasetAction,
	triggers []DatasetTrigger,
	contentDeliveryRules []ContentDeliveryRule,
	versioningConfig *VersioningConfiguration,
	lateDataRules []LateDataRule,
) (*Dataset, error)

CreateDataset creates a new IoT Analytics dataset.

func (*InMemoryBackend) CreateDatasetContent 

func (b *InMemoryBackend) CreateDatasetContent(datasetName, versionID string) (*DatasetContent, error)

CreateDatasetContent creates a new content version for a dataset. If versionID is non-empty, it is used as the new version's VersionID instead of generating a random one (AWS docs: "The version ID of the dataset content. To specify versionId for a dataset content, the dataset must use a DeltaTimer filter" -- this backend accepts an explicit versionId unconditionally rather than requiring a DeltaTimer trigger, since enforcing that restriction would require simulating DeltaTimer-driven content generation this backend does not otherwise model). A duplicate explicit versionID is rejected with ErrAlreadyExists rather than silently overwriting an existing content version.

func (*InMemoryBackend) CreateDatastore 

func (b *InMemoryBackend) CreateDatastore(
	ctx context.Context,
	name string,
	tags map[string]string,
	storage *DatastoreStorage,
	retention *RetentionPeriod,
	fileFormat *FileFormatConfiguration,
	partitions *DatastorePartitions,
) (*Datastore, error)

CreateDatastore creates a new IoT Analytics datastore.

func (*InMemoryBackend) CreatePipeline 

func (b *InMemoryBackend) CreatePipeline(
	ctx context.Context,
	name string,
	tags map[string]string,
	activities []PipelineActivity,
) (*Pipeline, error)

CreatePipeline creates a new IoT Analytics pipeline.

func (*InMemoryBackend) DeleteChannel 

func (b *InMemoryBackend) DeleteChannel(name string) error

DeleteChannel deletes a channel and its associated messages.

func (*InMemoryBackend) DeleteDataset 

func (b *InMemoryBackend) DeleteDataset(name string) error

DeleteDataset deletes a dataset and its associated content versions.

func (*InMemoryBackend) DeleteDatasetContent 

func (b *InMemoryBackend) DeleteDatasetContent(datasetName, versionID string) error

DeleteDatasetContent deletes a single content version, matching AWS DeleteDatasetContent versionId semantics: a specific versionId, $LATEST (the most recently created version regardless of status), or $LATEST_SUCCEEDED (the default when versionID is empty -- the most recently created SUCCEEDED version). Unlike an unqualified "delete all", AWS never removes more than one content version per call.

func (*InMemoryBackend) DeleteDatastore 

func (b *InMemoryBackend) DeleteDatastore(name string) error

DeleteDatastore deletes a datastore.

func (*InMemoryBackend) DeletePipeline 

func (b *InMemoryBackend) DeletePipeline(name string) error

DeletePipeline deletes a pipeline.

func (*InMemoryBackend) DescribeChannel 

func (b *InMemoryBackend) DescribeChannel(name string) (*Channel, error)

DescribeChannel returns channel metadata.

func (*InMemoryBackend) DescribeDataset 

func (b *InMemoryBackend) DescribeDataset(name string) (*Dataset, error)

DescribeDataset returns dataset metadata.

func (*InMemoryBackend) DescribeDatastore 

func (b *InMemoryBackend) DescribeDatastore(name string) (*Datastore, error)

DescribeDatastore returns datastore metadata.

func (*InMemoryBackend) DescribeLoggingOptions 

func (b *InMemoryBackend) DescribeLoggingOptions() (*LoggingOptions, error)

DescribeLoggingOptions returns the current IoT Analytics logging options.

func (*InMemoryBackend) DescribePipeline 

func (b *InMemoryBackend) DescribePipeline(name string) (*Pipeline, error)

DescribePipeline returns pipeline metadata.

func (*InMemoryBackend) GetDatasetContent 

func (b *InMemoryBackend) GetDatasetContent(datasetName, versionID string) (*DatasetContent, error)

GetDatasetContent retrieves a specific, latest ($LATEST), or latest-succeeded ($LATEST_SUCCEEDED, also the default when versionID is empty) content version of a dataset, matching AWS GetDatasetContent versionId semantics.

func (*InMemoryBackend) ListChannels 

func (b *InMemoryBackend) ListChannels() []*Channel

ListChannels returns all channels sorted by name.

func (*InMemoryBackend) ListDatasetContents 

func (b *InMemoryBackend) ListDatasetContents(datasetName string) ([]*DatasetContent, error)

ListDatasetContents returns all content versions for a dataset, sorted by creation time descending, ties broken by insertion order (most recently created first). CreationTime has only second-level resolution (epochSeconds), so content versions created within the same test or request burst routinely tie; a plain slices.SortFunc is explicitly documented as unstable and would then reorder tied entries arbitrarily between calls, which breaks the offset-based pagination in handleListDatasetContents (two calls for page 1 and page 2 could disagree on ordering with nothing mutated in between). Reversing to newest-inserted-first before a *stable* sort makes ties resolve deterministically in that same direction.

func (*InMemoryBackend) ListDatasets 

func (b *InMemoryBackend) ListDatasets() []*Dataset

ListDatasets returns all datasets sorted by name.

func (*InMemoryBackend) ListDatastores 

func (b *InMemoryBackend) ListDatastores() []*Datastore

ListDatastores returns all datastores sorted by name.

func (*InMemoryBackend) ListPipelines 

func (b *InMemoryBackend) ListPipelines() []*Pipeline

ListPipelines returns all pipelines sorted by name.

func (*InMemoryBackend) ListTagsForResource 

func (b *InMemoryBackend) ListTagsForResource(resourceARN string) ([]TagDTO, error)

ListTagsForResource returns tags for a resource ARN, sorted by key. Returns empty slice (not error) when the resource exists but has no tags.

func (*InMemoryBackend) PutLoggingOptions 

func (b *InMemoryBackend) PutLoggingOptions(options *LoggingOptions) error

PutLoggingOptions sets the IoT Analytics logging options. Validates: level must be "ERROR"; roleArn is required when enabled is true.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all backend state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore deserializes backend state from a JSON snapshot.

func (*InMemoryBackend) RunPipelineActivity 

func (b *InMemoryBackend) RunPipelineActivity(activity PipelineActivity, payloads [][]byte) ([][]byte, error)

RunPipelineActivity runs payloads through a single pipeline activity and returns the results, matching AWS RunPipelineActivity semantics per activity type:

  • addAttributes/removeAttributes/selectAttributes: pure JSON-object transforms, applied to every payload that parses as a JSON object (see applyAddAttributes et al.).
  • filter: evaluates the SQL-like filter expression (see pipeline_expr.go) against each payload and returns only the payloads that match -- non-matching or unparsable payloads are dropped from the pipeline, exactly as a real filter activity would.
  • math: evaluates the math expression and stores the result under Attribute.
  • channel/datastore: legitimately pass-through in real AWS too (they are the pipeline's source/sink activities, not transforms).
  • lambda/deviceRegistryEnrich/deviceShadowEnrich: real AWS invokes Lambda / looks up IoT Device Registry or Device Shadow data for these. This backend has no cross-service wiring to reach the lambda/iot backends from iotanalytics's Provider.Init (no shared backend registry is threaded through, unlike e.g. cloudformation's ResourceCreator), so these remain pass-through -- a documented gap, not a silent stub (see PARITY.md).

A payload that fails to parse as JSON, or a message missing a referenced attribute, is a soft per-message failure: it is left unchanged (addAttributes/removeAttributes/ selectAttributes/math) or dropped (filter), matching a single bad message failing its own pipeline activity step rather than the entire batch call.

func (*InMemoryBackend) SampleChannelData 

func (b *InMemoryBackend) SampleChannelData(channelName string, maxMessages int) ([][]byte, error)

SampleChannelData returns up to maxMessages sample messages from a channel. Returns InvalidRequestException for maxMessages <= 0 or > 10 (AWS behaviour).

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serializes backend state to JSON.

func (*InMemoryBackend) StartPipelineReprocessing 

func (b *InMemoryBackend) StartPipelineReprocessing(pipelineName string, startTime, endTime *float64) (string, error)

StartPipelineReprocessing creates a new reprocessing job for a pipeline. Optional startTime and endTime define the message window to reprocess.

func (*InMemoryBackend) TagResource 

func (b *InMemoryBackend) TagResource(resourceARN string, tags []TagDTO) error

TagResource adds or updates tags on a resource, enforcing the per-resource tag limit.

func (*InMemoryBackend) UntagResource 

func (b *InMemoryBackend) UntagResource(resourceARN string, tagKeys []string) error

UntagResource removes tags from a resource.

func (*InMemoryBackend) UpdateChannel 

func (b *InMemoryBackend) UpdateChannel(name string, storage *ChannelStorage, retention *RetentionPeriod) error

UpdateChannel updates a channel's storage configuration, retention period, and last update time.

func (*InMemoryBackend) UpdateDataset 

func (b *InMemoryBackend) UpdateDataset(
	name string,
	actions []DatasetAction,
	triggers []DatasetTrigger,
	contentDeliveryRules []ContentDeliveryRule,
	versioningConfig *VersioningConfiguration,
	lateDataRules []LateDataRule,
) error

UpdateDataset updates a dataset's actions, triggers, and configuration.

func (*InMemoryBackend) UpdateDatastore 

func (b *InMemoryBackend) UpdateDatastore(
	name string,
	storage *DatastoreStorage,
	retention *RetentionPeriod,
	fileFormat *FileFormatConfiguration,
	partitions *DatastorePartitions,
) error

UpdateDatastore updates a datastore's configuration and last update time.

func (*InMemoryBackend) UpdatePipeline 

func (b *InMemoryBackend) UpdatePipeline(name string, activities []PipelineActivity) error

UpdatePipeline updates a pipeline's activities and last update time.

type IotEventsDestination 

type IotEventsDestination struct {
	InputName string `json:"inputName"`
	RoleArn   string `json:"roleArn"`
}

IotEventsDestination delivers content to IoT Events.

type IotSiteWiseMultiLayerStorage 

type IotSiteWiseMultiLayerStorage struct {
	CustomerManagedS3Storage *CustomerManagedS3DatastoreStorage `json:"customerManagedS3Storage,omitempty"`
}

IotSiteWiseMultiLayerStorage is IoT SiteWise multi-layer storage for datastores.

type JSONConfiguration 

type JSONConfiguration struct{}

JSONConfiguration defines JSON file format settings (marker type).

type LateDataRule 

type LateDataRule struct {
	RuleConfiguration *LateDataRuleConfiguration `json:"ruleConfiguration"`
	RuleName          string                     `json:"ruleName,omitempty"`
}

LateDataRule defines conditions under which late data triggers dataset refresh.

type LateDataRuleConfiguration 

type LateDataRuleConfiguration struct {
	DeltaTimeSessionWindowConfiguration *DeltaTimeSessionWindowConfiguration `json:"deltaTimeSessionWindowConfiguration,omitempty"` //nolint:lll // AWS field name
}

LateDataRuleConfiguration is the configuration for a late data rule.

type LoggingOptions 

type LoggingOptions struct {
	RoleARN string `json:"roleArn"`
	Level   string `json:"level"`
	Enabled bool   `json:"enabled"`
}

LoggingOptions stores the IoT Analytics logging configuration.

type ParquetConfiguration 

type ParquetConfiguration struct {
	SchemaDefinition *SchemaDefinition `json:"schemaDefinition,omitempty"`
}

ParquetConfiguration defines Parquet file format settings.

type Pipeline 

type Pipeline struct {
	Tags          map[string]string                `json:"tags"`
	Reprocessings map[string]*PipelineReprocessing `json:"reprocessings"`
	Name          string                           `json:"name"`
	ARN           string                           `json:"arn"`
	Activities    []PipelineActivity               `json:"activities,omitempty"`
	CreationTime  float64                          `json:"creationTime"`
	LastUpdate    float64                          `json:"lastUpdate"`
}

Pipeline stores all metadata and state for a single IoT Analytics pipeline.

type PipelineActivity 

type PipelineActivity struct {
	Channel              *PipelineChannelActivity              `json:"channel,omitempty"`
	Lambda               *PipelineLambdaActivity               `json:"lambda,omitempty"`
	Datastore            *PipelineDatastoreActivity            `json:"datastore,omitempty"`
	AddAttributes        *PipelineAddAttributesActivity        `json:"addAttributes,omitempty"`
	RemoveAttributes     *PipelineRemoveAttributesActivity     `json:"removeAttributes,omitempty"`
	SelectAttributes     *PipelineSelectAttributesActivity     `json:"selectAttributes,omitempty"`
	Filter               *PipelineFilterActivity               `json:"filter,omitempty"`
	Math                 *PipelineMathActivity                 `json:"math,omitempty"`
	DeviceRegistryEnrich *PipelineDeviceRegistryEnrichActivity `json:"deviceRegistryEnrich,omitempty"`
	DeviceShadowEnrich   *PipelineDeviceShadowEnrichActivity   `json:"deviceShadowEnrich,omitempty"`
}

PipelineActivity is a typed pipeline activity union.

type PipelineAddAttributesActivity 

type PipelineAddAttributesActivity struct {
	Attributes map[string]string `json:"attributes"`
	Name       string            `json:"name"`
	Next       string            `json:"next,omitempty"`
}

PipelineAddAttributesActivity adds attributes to messages.

type PipelineChannelActivity 

type PipelineChannelActivity struct {
	ChannelName string `json:"channelName"`
	Name        string `json:"name"`
	Next        string `json:"next,omitempty"`
}

PipelineChannelActivity is the pipeline channel source activity.

type PipelineDatastoreActivity 

type PipelineDatastoreActivity struct {
	DatastoreName string `json:"datastoreName"`
	Name          string `json:"name"`
}

PipelineDatastoreActivity is the pipeline sink activity.

type PipelineDeviceRegistryEnrichActivity 

type PipelineDeviceRegistryEnrichActivity struct {
	Attribute string `json:"attribute"`
	ThingName string `json:"thingName"`
	RoleArn   string `json:"roleArn"`
	Name      string `json:"name"`
	Next      string `json:"next,omitempty"`
}

PipelineDeviceRegistryEnrichActivity enriches messages with Device Registry data.

type PipelineDeviceShadowEnrichActivity 

type PipelineDeviceShadowEnrichActivity struct {
	Attribute string `json:"attribute"`
	ThingName string `json:"thingName"`
	RoleArn   string `json:"roleArn"`
	Name      string `json:"name"`
	Next      string `json:"next,omitempty"`
}

PipelineDeviceShadowEnrichActivity enriches messages with Device Shadow data.

type PipelineFilterActivity 

type PipelineFilterActivity struct {
	Filter string `json:"filter"`
	Name   string `json:"name"`
	Next   string `json:"next,omitempty"`
}

PipelineFilterActivity filters messages based on a condition.

type PipelineLambdaActivity 

type PipelineLambdaActivity struct {
	LambdaName string `json:"lambdaName"`
	Name       string `json:"name"`
	Next       string `json:"next,omitempty"`
	BatchSize  int    `json:"batchSize,omitempty"`
}

PipelineLambdaActivity invokes a Lambda function on messages.

type PipelineMathActivity 

type PipelineMathActivity struct {
	Attribute string `json:"attribute"`
	Math      string `json:"math"`
	Name      string `json:"name"`
	Next      string `json:"next,omitempty"`
}

PipelineMathActivity computes a math expression and adds result as an attribute.

type PipelineRemoveAttributesActivity 

type PipelineRemoveAttributesActivity struct {
	Name       string   `json:"name"`
	Next       string   `json:"next,omitempty"`
	Attributes []string `json:"attributes"`
}

PipelineRemoveAttributesActivity removes attributes from messages.

type PipelineReprocessing 

type PipelineReprocessing struct {
	ID           string  `json:"id"`
	Status       string  `json:"status"`
	CreationTime float64 `json:"creationTime"`
	EndTime      float64 `json:"endTime,omitempty"`
	StartTime    float64 `json:"startTime,omitempty"`
}

PipelineReprocessing stores state for a single pipeline reprocessing job.

type PipelineSelectAttributesActivity 

type PipelineSelectAttributesActivity struct {
	Name       string   `json:"name"`
	Next       string   `json:"next,omitempty"`
	Attributes []string `json:"attributes"`
}

PipelineSelectAttributesActivity selects specific attributes from messages.

type Provider 

type Provider struct{}

Provider implements service.Provider for the IoT Analytics service.

func (*Provider) Init 

func (p *Provider) Init(appCtx *service.AppContext) (service.Registerable, error)

Init initializes the IoT Analytics service backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the provider name.

type ResourceConfiguration 

type ResourceConfiguration struct {
	ComputeType    string `json:"computeType"`
	VolumeSizeInGB int    `json:"volumeSizeInGB"`
}

ResourceConfiguration defines compute resources for container actions.

type RetentionPeriod 

type RetentionPeriod struct {
	NumberOfDays int  `json:"numberOfDays,omitempty"`
	Unlimited    bool `json:"unlimited,omitempty"`
}

RetentionPeriod defines how long data is retained. Exactly one of Unlimited or NumberOfDays must be set.

type S3DestinationConfiguration 

type S3DestinationConfiguration struct {
	GlueConfiguration *GlueConfiguration `json:"glueConfiguration,omitempty"`
	Bucket            string             `json:"bucket"`
	Key               string             `json:"key"`
	RoleArn           string             `json:"roleArn"`
}

S3DestinationConfiguration delivers content to S3.

type ScheduleExpression 

type ScheduleExpression struct {
	Expression string `json:"expression"`
}

ScheduleExpression defines a cron-based schedule trigger.

type SchemaDefinition 

type SchemaDefinition struct {
	Columns []ColumnSchema `json:"columns"`
}

SchemaDefinition defines the schema for Parquet format.

type ServiceManagedS3Storage 

type ServiceManagedS3Storage struct{}

ServiceManagedS3Storage indicates AWS-managed S3 storage (marker type).

type Snapshottable 

type Snapshottable interface {
	Snapshot(ctx context.Context) []byte
	Restore(context.Context, []byte) error
}

Snapshottable is an optional interface that a StorageBackend may implement to support snapshot/restore for persistence or test isolation.

type StorageBackend 

type StorageBackend interface {
	CreateChannel(
		ctx context.Context,
		name string,
		tags map[string]string,
		storage *ChannelStorage,
		retention *RetentionPeriod,
	) (*Channel, error)
	DescribeChannel(name string) (*Channel, error)
	UpdateChannel(name string, storage *ChannelStorage, retention *RetentionPeriod) error
	DeleteChannel(name string) error
	ListChannels() []*Channel

	CreateDatastore(
		ctx context.Context,
		name string,
		tags map[string]string,
		storage *DatastoreStorage,
		retention *RetentionPeriod,
		fileFormat *FileFormatConfiguration,
		partitions *DatastorePartitions,
	) (*Datastore, error)
	DescribeDatastore(name string) (*Datastore, error)
	UpdateDatastore(
		name string,
		storage *DatastoreStorage,
		retention *RetentionPeriod,
		fileFormat *FileFormatConfiguration,
		partitions *DatastorePartitions,
	) error
	DeleteDatastore(name string) error
	ListDatastores() []*Datastore

	CreateDataset(
		ctx context.Context,
		name string,
		tags map[string]string,
		actions []DatasetAction,
		triggers []DatasetTrigger,
		contentDeliveryRules []ContentDeliveryRule,
		versioningConfig *VersioningConfiguration,
		lateDataRules []LateDataRule,
	) (*Dataset, error)
	DescribeDataset(name string) (*Dataset, error)
	UpdateDataset(
		name string,
		actions []DatasetAction,
		triggers []DatasetTrigger,
		contentDeliveryRules []ContentDeliveryRule,
		versioningConfig *VersioningConfiguration,
		lateDataRules []LateDataRule,
	) error
	DeleteDataset(name string) error
	ListDatasets() []*Dataset

	CreatePipeline(
		ctx context.Context,
		name string,
		tags map[string]string,
		activities []PipelineActivity,
	) (*Pipeline, error)
	DescribePipeline(name string) (*Pipeline, error)
	UpdatePipeline(name string, activities []PipelineActivity) error
	DeletePipeline(name string) error
	ListPipelines() []*Pipeline

	ListTagsForResource(resourceARN string) ([]TagDTO, error)
	TagResource(resourceARN string, tags []TagDTO) error
	UntagResource(resourceARN string, tagKeys []string) error

	BatchPutMessage(channelName string, messages []messageInput) ([]BatchPutMessageErrorEntry, error)
	SampleChannelData(channelName string, maxMessages int) ([][]byte, error)

	StartPipelineReprocessing(pipelineName string, startTime, endTime *float64) (string, error)
	CancelPipelineReprocessing(pipelineName, reprocessingID string) error

	CreateDatasetContent(datasetName, versionID string) (*DatasetContent, error)
	GetDatasetContent(datasetName, versionID string) (*DatasetContent, error)
	ListDatasetContents(datasetName string) ([]*DatasetContent, error)
	DeleteDatasetContent(datasetName, versionID string) error

	DescribeLoggingOptions() (*LoggingOptions, error)
	PutLoggingOptions(options *LoggingOptions) error

	RunPipelineActivity(activity PipelineActivity, payloads [][]byte) ([][]byte, error)

	Reset()
}

StorageBackend is the interface for the IoT Analytics backend.

type TagDTO 

type TagDTO struct {
	Key   string `json:"key"`
	Value string `json:"value"`
}

TagDTO is a key-value tag for IoT Analytics resources.

type TimestampPartition 

type TimestampPartition struct {
	AttributeName   string `json:"attributeName"`
	TimestampFormat string `json:"timestampFormat,omitempty"`
}

TimestampPartition defines a datastore partition by timestamp attribute.

type VersioningConfiguration 

type VersioningConfiguration struct {
	MaxVersions int  `json:"maxVersions,omitempty"`
	Unlimited   bool `json:"unlimited,omitempty"`
}

VersioningConfiguration controls how many content versions to retain.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.