delivery

Documentation

Index

• Variables
• func Deliver(ctx context.Context, id string, path string, s Source, d Destination) (string, error)
• func GetDeliveredFilename(ctx context.Context, tuid string, pathTemplate string, ...) (string, error)
• func RegisterSource(name string, s Source)
• type AzureDestination
  • func (ad *AzureDestination) Client() (*container.Client, error)
  • func (ad *AzureDestination) Health(ctx context.Context) (rsp models.ServiceHealthResp)
  • func (ad *AzureDestination) Upload(ctx context.Context, path string, r io.Reader, m map[string]string) (string, error)
• type AzureSource
  • func (ad *AzureSource) GetMetadata(ctx context.Context, tuid string) (map[string]string, error)
  • func (ad *AzureSource) Health(ctx context.Context) (rsp models.ServiceHealthResp)
  • func (ad *AzureSource) Reader(ctx context.Context, path string) (io.Reader, error)
• type Config
  • func UnmarshalDeliveryConfig(confBody string) (*Config, error)
• type Destination
  • func GetTarget(target string) (Destination, bool)
• type FileDestination
  • func (fd *FileDestination) Health(_ context.Context) (rsp models.ServiceHealthResp)
  • func (fd *FileDestination) Upload(ctx context.Context, path string, r io.Reader, m map[string]string) (string, error)
• type FileSource
  • func (fd *FileSource) GetMetadata(_ context.Context, tuid string) (map[string]string, error)
  • func (fd *FileSource) Health(_ context.Context) (rsp models.ServiceHealthResp)
  • func (fd *FileSource) Reader(_ context.Context, path string) (io.Reader, error)
• type Group
  • func FindGroupFromMetadata(meta handler.MetaData) (Group, bool)
  • func (g *Group) Key() string
  • func (g *Group) TargetNames() []string
• type PathInfo
• type S3Destination
  • func (sd *S3Destination) Client() *s3.Client
  • func (sd *S3Destination) Health(ctx context.Context) (rsp models.ServiceHealthResp)
  • func (sd *S3Destination) Retrieve(_ context.Context) (aws.Credentials, error)
  • func (sd *S3Destination) Upload(ctx context.Context, path string, r io.Reader, m map[string]string) (string, error)
• type S3Source
  • func (ss *S3Source) GetMetadata(ctx context.Context, id string) (map[string]string, error)
  • func (ss *S3Source) Health(ctx context.Context) (rsp models.ServiceHealthResp)
  • func (ss *S3Source) Reader(ctx context.Context, id string) (io.Reader, error)
• type Source
  • func GetSource(name string) (Source, bool)
• type Target
  • func (t *Target) UnmarshalYAML(n *yaml.Node) error
• type TargetDesignation

Variables

var DestinationTypes = map[string]func() Destination{
	"s3":      func() Destination { return &S3Destination{} },
	"file":    func() Destination { return &FileDestination{} },
	"az-blob": func() Destination { return &AzureDestination{} },
}

var ErrBadIngestTimestamp = errors.New("bad ingest timestamp")

var ErrSrcFileNotExist = fmt.Errorf("source file does not exist")

var ErrUnknownDestinationType = errors.New("Unknown destination type")

var Groups map[string]Group

var Targets map[string]Destination

var UploadSrc = "upload"

Functions

func Deliver

func Deliver(ctx context.Context, id string, path string, s Source, d Destination) (string, error)

target may end up being a type

func GetDeliveredFilename

func GetDeliveredFilename(ctx context.Context, tuid string, pathTemplate string, manifest map[string]string) (string, error)

func RegisterSource

func RegisterSource(name string, s Source)

Types

type AzureDestination

type AzureDestination struct {
	Name              string `yaml:"name"`
	StorageAccount    string `yaml:"storage_account"`
	StorageKey        string `yaml:"storage_key"`
	PathTemplate      string `yaml:"path_template"`
	ContainerEndpoint string `yaml:"endpoint"`
	TenantId          string `yaml:"tenant_id"`
	ClientId          string `yaml:"client_id"`
	ClientSecret      string `yaml:"client_secret"`
	ContainerName     string `yaml:"container_name"`
	// contains filtered or unexported fields
}

func (*AzureDestination) Client

func (ad *AzureDestination) Client() (*container.Client, error)

func (*AzureDestination) Health

func (ad *AzureDestination) Health(ctx context.Context) (rsp models.ServiceHealthResp)

func (*AzureDestination) Upload

func (ad *AzureDestination) Upload(ctx context.Context, path string, r io.Reader, m map[string]string) (string, error)

type AzureSource

type AzureSource struct {
	FromContainerClient *container.Client
	Prefix              string
}

func (*AzureSource) GetMetadata

func (ad *AzureSource) GetMetadata(ctx context.Context, tuid string) (map[string]string, error)

func (*AzureSource) Health

func (ad *AzureSource) Health(ctx context.Context) (rsp models.ServiceHealthResp)

func (*AzureSource) Reader

func (ad *AzureSource) Reader(ctx context.Context, path string) (io.Reader, error)

type Config

type Config struct {
	Targets map[string]Target `yaml:"targets"`
	Groups  []Group           `yaml:"routing_groups"`
}

func UnmarshalDeliveryConfig

func UnmarshalDeliveryConfig(confBody string) (*Config, error)

type Destination

type Destination interface {
	Upload(context.Context, string, io.Reader, map[string]string) (string, error)
}

func GetTarget

func GetTarget(target string) (Destination, bool)

type FileDestination

type FileDestination struct {
	ToPath       string `yaml:"path"`
	Name         string `yaml:"name"`
	PathTemplate string `yaml:"path_template"`
}

func (*FileDestination) Health

func (fd *FileDestination) Health(_ context.Context) (rsp models.ServiceHealthResp)

func (*FileDestination) Upload

func (fd *FileDestination) Upload(ctx context.Context, path string, r io.Reader, m map[string]string) (string, error)

type FileSource

type FileSource struct {
	FS fs.FS
}

func (*FileSource) GetMetadata

func (fd *FileSource) GetMetadata(_ context.Context, tuid string) (map[string]string, error)

func (*FileSource) Health

func (fd *FileSource) Health(_ context.Context) (rsp models.ServiceHealthResp)

func (*FileSource) Reader

func (fd *FileSource) Reader(_ context.Context, path string) (io.Reader, error)

type Group

type Group struct {
	DataStreamId    string              `yaml:"data_stream_id"`
	DataStreamRoute string              `yaml:"data_stream_route"`
	DeliveryTargets []TargetDesignation `yaml:"delivery_targets"`
}

func FindGroupFromMetadata

func FindGroupFromMetadata(meta handler.MetaData) (Group, bool)

func (*Group) Key

func (g *Group) Key() string

func (*Group) TargetNames

func (g *Group) TargetNames() []string

type PathInfo

type PathInfo struct {
	Year            string
	Month           string
	Day             string
	Hour            string
	UploadId        string
	Filename        string
	Suffix          string
	Prefix          string
	DataStreamID    string
	DataStreamRoute string
}

type S3Destination

type S3Destination struct {
	BucketName      string `yaml:"bucket_name"`
	Name            string `yaml:"name"`
	PathTemplate    string `yaml:"path_template"`
	AccessKeyID     string `yaml:"access_key_id"`
	SecretAccessKey string `yaml:"secret_access_key"`
	Endpoint        string `yaml:"endpoint"`
	Region          string `yaml:"region"`
	// contains filtered or unexported fields
}

func (*S3Destination) Client

func (sd *S3Destination) Client() *s3.Client

func (*S3Destination) Health

func (sd *S3Destination) Health(ctx context.Context) (rsp models.ServiceHealthResp)

func (*S3Destination) Retrieve

func (sd *S3Destination) Retrieve(_ context.Context) (aws.Credentials, error)

func (*S3Destination) Upload

func (sd *S3Destination) Upload(ctx context.Context, path string, r io.Reader, m map[string]string) (string, error)

type S3Source

type S3Source struct {
	FromClient *s3.Client
	BucketName string
	Prefix     string
}

func (*S3Source) GetMetadata

func (ss *S3Source) GetMetadata(ctx context.Context, id string) (map[string]string, error)

func (*S3Source) Health

func (ss *S3Source) Health(ctx context.Context) (rsp models.ServiceHealthResp)

func (*S3Source) Reader

func (ss *S3Source) Reader(ctx context.Context, id string) (io.Reader, error)

type Source

type Source interface {
	Reader(context.Context, string) (io.Reader, error)
	GetMetadata(context.Context, string) (map[string]string, error)
}

func GetSource

func GetSource(name string) (Source, bool)

type Target

type Target struct {
	Name        string      `yaml:"name"`
	Type        string      `yaml:"type"`
	Destination Destination `yaml:"-"`
}

func (*Target) UnmarshalYAML

func (t *Target) UnmarshalYAML(n *yaml.Node) error

type TargetDesignation

type TargetDesignation struct {
	Name         string `yaml:"name"`
	PathTemplate string `yaml:"path_template"`
}