ew_api

Documentation

Index

• Constants
• type ExportWorker
  • func NewExportWorker(kafkaProducer *kafka.Writer, kafkaConn *kafka.Conn, ...) *ExportWorker
  • func (ew *ExportWorker) CreateFilterTopic(topic string, checkExists bool) (err error)
  • func (ew *ExportWorker) CreateInstance(instance *lib.Instance, dataFields string, tagFields string) (serviceId string, err error)
  • func (ew *ExportWorker) DeleteInstance(instance *lib.Instance) (err error)
  • func (ew *ExportWorker) InitFilterTopics(serving *lib.Serving) (err error)
• type Filter
• type Identifier
• type InfluxDBExportArgs
• type Message
• type TimescaleDBExportArgs

Constants

const (
	MethodPut    = "put"
	MethodDelete = "delete"
)

const (
	TypeDevice       = "deviceId"
	TypeAnalytics    = "operatorId"
	TypeImport       = "import_id"
	MappingData      = ":data"
	MappingExtra     = ":extra"
	IdentKeyDevice   = "device_id"
	IdentKeyService  = "service_id"
	IdentKeyPipeline = "pipeline_id"
	IdentKeyOperator = "operator_id"
	IdentKeyImport   = "import_id"
	InfluxDB         = "influxdb"
	TimescaleDB      = "timescaledb"
)

const (
	InfluxDBTimeKey = "time"
)

Types

type ExportWorker

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

func NewExportWorker

func NewExportWorker(kafkaProducer *kafka.Writer, kafkaConn *kafka.Conn, kafkaControllerConn *kafka.Conn) *ExportWorker

func (*ExportWorker) CreateFilterTopic

func (ew *ExportWorker) CreateFilterTopic(topic string, checkExists bool) (err error)

func (*ExportWorker) CreateInstance

func (ew *ExportWorker) CreateInstance(instance *lib.Instance, dataFields string, tagFields string) (serviceId string, err error)

func (*ExportWorker) DeleteInstance

func (ew *ExportWorker) DeleteInstance(instance *lib.Instance) (err error)

func (*ExportWorker) InitFilterTopics

func (ew *ExportWorker) InitFilterTopics(serving *lib.Serving) (err error)

type Filter

type Filter struct {
	Source      string            `json:"source,omitempty"`
	Identifiers []Identifier      `json:"identifiers,omitempty"`
	Mappings    map[string]string `json:"mappings,omitempty"`
	ID          string            `json:"id"`
	Args        interface{}       `json:"args,omitempty"`
}

type Identifier

type Identifier struct {
	Key   string `json:"key"`
	Value string `json:"value,omitempty"`
}

type InfluxDBExportArgs

type InfluxDBExportArgs struct {
	DBName        string            `json:"db_name"`
	TypeCasts     map[string]string `json:"type_casts,omitempty"`
	TimeKey       string            `json:"time_key,omitempty"`
	TimeFormat    string            `json:"time_format,omitempty"`
	TimePrecision string            `json:"time_precision,omitempty"`
}

type Message

type Message struct {
	Method    string `json:"method"`
	Payload   Filter `json:"payload"`
	Timestamp int64  `json:"timestamp"`
}

type TimescaleDBExportArgs

type TimescaleDBExportArgs struct {
	TableName    string      `json:"table_name"`
	TableColumns [][3]string `json:"table_columns"`
	TimeColumn   string      `json:"time_column"`
	TimeFormat   string      `json:"time_format"`
	TimeUnique   bool        `json:"time_unique"`
}