storage

agentdata_dynamodb_storage.go

type DynamoDBAPI interface {
	GetItem(*dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error)
	Scan(*dynamodb.ScanInput) (*dynamodb.ScanOutput, error)
	PutItem(*dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error)
	UpdateItem(input *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error)
	DeleteItem(input *dynamodb.DeleteItemInput) (*dynamodb.DeleteItemOutput, error)
}

type DynamoDBStorage struct {
	client DynamoDBAPI
	table  string
}

type Option func(storage *DynamoDBStorage) error

func NewDynamoDBStorage(client DynamoDBAPI, options ...Option) (*DynamoDBStorage, error) {
	ddbs := &DynamoDBStorage{
		table: TableName,
	}

	for _, opt := range options {
		if opt == nil {
			continue
		}

		err := opt(ddbs)
		if err != nil {
			return nil, err
		}
	}

	ddbs.client = client

	return ddbs, nil
}

func WithTableName(name string) Option {
	return func(d *DynamoDBStorage) error {
		d.table = name

		log.Infof("dynamodb storage table set to '%s'", d.table)

		return nil
	}
}

func (ddbs *DynamoDBStorage) Get(id, skey string) (AgentData, error) {
	cd := AgentData{}

	if err := IDIsValid(id); err != nil {
		return cd, err
	}

	output, err := ddbs.client.GetItem(&dynamodb.GetItemInput{
		Key: map[string]*dynamodb.AttributeValue{
			"ID":   {S: aws.String(id)},
			"skey": {S: aws.String(skey)},
		},
		TableName: aws.String(ddbs.table),
	})
	if err != nil {
		return cd, err
	}

	if output.Item == nil {
		return cd, ErrNotFound
	}

	err = dynamodbattribute.Unmarshal(output.Item["data"], &cd)
	if err != nil {
		return cd, err
	}

	return cd, nil
}

func (ddbs *DynamoDBStorage) List(skey) ([]AgentData, error) {
	filt := expression.Name("skey").Equal(expression.Value(skey))

	expr, err := expression.NewBuilder().WithFilter(filt).Build()
	if err != nil {
		return []AgentData{}, err
	}

	outputs, err := ddbs.client.Scan(&dynamodb.ScanInput{
		TableName:                 aws.String(ddbs.table),
		ExpressionAttributeNames:  expr.Names(),
		ExpressionAttributeValues: expr.Values(),
		FilterExpression:          expr.Filter(),
	})
	if err != nil {
		return []AgentData{}, err
	}

	entries := make([]AgentData, 0, len(outputs.Items))

	for _, item := range outputs.Items {
		var cd AgentData

		err = dynamodbattribute.Unmarshal(item["data"], &cd)
		if err != nil {
			log.Errorf("error unmarshalling deployment: %s", *(item["ID"].S))

			continue
		}

		entries = append(entries, cd)
	}

	return entries, nil
}

customdata_dynamodb_storage.go

type DynamoDBAPI interface {
	GetItem(*dynamodb.GetItemInput) (*dynamodb.GetItemOutput, error)
	Scan(*dynamodb.ScanInput) (*dynamodb.ScanOutput, error)
	PutItem(*dynamodb.PutItemInput) (*dynamodb.PutItemOutput, error)
	UpdateItem(input *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error)
	DeleteItem(input *dynamodb.DeleteItemInput) (*dynamodb.DeleteItemOutput, error)
}

type DynamoDBStorage struct {
	client DynamoDBAPI
	table  string
}

type Option func(storage *DynamoDBStorage) error

func NewDynamoDBStorage(client DynamoDBAPI, options ...Option) (*DynamoDBStorage, error) {
	ddbs := &DynamoDBStorage{
		table: TableName,
	}

	for _, opt := range options {
		if opt == nil {
			continue
		}

		err := opt(ddbs)
		if err != nil {
			return nil, err
		}
	}

	ddbs.client = client

	return ddbs, nil
}

func WithTableName(name string) Option {
	return func(d *DynamoDBStorage) error {
		d.table = name

		log.Infof("dynamodb storage table set to '%s'", d.table)

		return nil
	}
}

func (ddbs *DynamoDBStorage) Get(id, skey string) (CustomData, error) {
	cd := CustomData{}

	if err := IDIsValid(id); err != nil {
		return cd, err
	}

	output, err := ddbs.client.GetItem(&dynamodb.GetItemInput{
		Key: map[string]*dynamodb.AttributeValue{
			"ID":   {S: aws.String(id)},
			"skey": {S: aws.String(skey)},
		},
		TableName: aws.String(ddbs.table),
	})
	if err != nil {
		return cd, err
	}

	if output.Item == nil {
		return cd, ErrNotFound
	}

	err = dynamodbattribute.Unmarshal(output.Item["data"], &cd)
	if err != nil {
		return cd, err
	}

	return cd, nil
}

func (ddbs *DynamoDBStorage) List(skey) ([]CustomData, error) {
	filt := expression.Name("skey").Equal(expression.Value(skey))

	expr, err := expression.NewBuilder().WithFilter(filt).Build()
	if err != nil {
		return []CustomData{}, err
	}

	outputs, err := ddbs.client.Scan(&dynamodb.ScanInput{
		TableName:                 aws.String(ddbs.table),
		ExpressionAttributeNames:  expr.Names(),
		ExpressionAttributeValues: expr.Values(),
		FilterExpression:          expr.Filter(),
	})
	if err != nil {
		return []CustomData{}, err
	}

	entries := make([]CustomData, 0, len(outputs.Items))

	for _, item := range outputs.Items {
		var cd CustomData

		err = dynamodbattribute.Unmarshal(item["data"], &cd)
		if err != nil {
			log.Errorf("error unmarshalling deployment: %s", *(item["ID"].S))

			continue
		}

		entries = append(entries, cd)
	}

	return entries, nil
}