config

Documentation

Index

• Constants
• func Environment() string
• func Load() error
• func LoadSection(cfg interface{}) error
• func Save() error
• func Set(a *AppConfigODS) error
• func Zone() string
• type AppConfig
  • func Get() *AppConfig
  • func (c *AppConfig) GetChangelogTopicName(svc string, db string, tbl string, input string, output string, ver int, ...) (string, error)
  • func (c *AppConfig) GetOutputTopicName(svc string, db string, tbl string, input string, output string, ver int, ...) (string, error)
• type AppConfigODS
• type EncryptionConfig
  • func (e EncryptionConfig) String() string
• type HadoopConfig
• type KafkaConfig
• type PipeConfig
• type RowFilter
• type S3Config
• type SQLConfig
• type ScheduleConfig
• type TableParams
  • func (t *TableParams) CopyForMerge() *TableParams
  • func (t *TableParams) MergeCompound(r *TableParams)

Constants

const (
	EnvProduction  = "production"
	EnvDevelopment = "development"
	EnvStaging     = "staging"
	EnvTest        = "test"
)

Define various environments

Functions

func Environment

func Environment() string

Environment returns current environment

func Load

func Load() error

Load creates the config

func LoadSection

func LoadSection(cfg interface{}) error

LoadSection can be used to load subsections of config files at runtime

func Save

func Save() error

Save saves config to backends

func Set

func Set(a *AppConfigODS) error

Set replaces current config with the contents of parameter

func Zone

func Zone() string

Zone returns the current zone that the application is running in

Types

type AppConfig

type AppConfig struct {
	AppConfigODS
	PortDyn                                 int
	ChangelogTopicNameTemplateParsed        map[string]map[string]*template.Template
	OutputTopicNameTemplateParsed           map[string]map[string]*template.Template
	ChangelogTopicNameTemplateDefaultParsed *template.Template
	OutputTopicNameTemplateDefaultParsed    *template.Template
}

AppConfig is the config struct which the config gets loaded into

func Get

func Get() *AppConfig

Get returns current config. Reloads it if refersh interval is expired

func (*AppConfig) GetChangelogTopicName

func (c *AppConfig) GetChangelogTopicName(svc string, db string, tbl string, input string, output string, ver int, ts time.Time) (string, error)

GetChangelogTopicName returns output topic name

func (*AppConfig) GetOutputTopicName

func (c *AppConfig) GetOutputTopicName(svc string, db string, tbl string, input string, output string, ver int, ts time.Time) (string, error)

GetOutputTopicName returns output topic name

type AppConfigODS

type AppConfigODS struct {
	LogType  string `yaml:"log_type"`
	LogLevel string `yaml:"log_level"`

	ServiceName string `yaml:"serviceName"`
	Port        int    `yaml:"port"`
	MaxNumProcs int    `yaml:"max_num_procs"`

	StateUpdateInterval time.Duration `yaml:"state_update_interval"`
	WorkerIdleInterval  time.Duration `yaml:"worker_idle_interval"`
	LockExpireTimeout   time.Duration `yaml:"lock_expire_timeout"`
	StateConnectURL     string        `yaml:"state_connect_url"`
	StateDBName         string        `yaml:"state_db_name"`
	StateClusterName    string        `yaml:"state_cluster_name"`

	ChangelogPipeType                 string                       `yaml:"changelog_pipe_type"`
	ChangelogTopicNameTemplateDefault string                       `yaml:"changelog_topic_name_template_default"`
	ChangelogTopicNameTemplate        map[string]map[string]string `yaml:"changelog_topic_name_template"`
	ChangelogWatchdogInterval         time.Duration

	OutputTopicNameTemplateDefault string                       `yaml:"output_topic_name_template_default"`
	OutputTopicNameTemplate        map[string]map[string]string `yaml:"output_topic_name_template"`

	Verbose bool

	ChangelogBuffer       bool `yaml:"changelog_buffer"`
	ForceMasterConnection bool `yaml:"force_master_connection"`

	InternalEncoding string `yaml:"internal_encoding"`

	TableParams `yaml:",inline"` //Merged with table specific config if any

	Filters map[string]map[string]RowFilter `yaml:"filters"`

	ConfigRefreshInterval time.Duration `yaml:"config_refresh_interval"`
}

AppConfigODS is the config struct which the config gets loaded into

type EncryptionConfig

type EncryptionConfig struct {
	Enabled    bool
	PublicKey  string `yaml:"public_key"`  // used to encrypt in producer
	PrivateKey string `yaml:"private_key"` // used to decrypt in consumer
	SigningKey string `yaml:"signing_key"` // used to sign in producer and verify in consumer
}

EncryptionConfig holds encryption configuration options

func (EncryptionConfig) String

func (e EncryptionConfig) String() string

String sanitizes config for log output

type HadoopConfig

type HadoopConfig struct {
	User      string
	Addresses []string
	BaseDir   string `yaml:"base_dir"`
}

HadoopConfig holds hadoop output pipe configuration

type KafkaConfig

type KafkaConfig struct {
	Addresses       []string
	MaxMessageBytes int `yaml:"max_message_bytes"`
}

KafkaConfig holds Kafka pipe configuration

type PipeConfig

type PipeConfig struct {
	MaxBatchSize      int `yaml:"max_batch_size"`
	MaxBatchSizeBytes int `yaml:"max_batch_size_bytes"`

	BaseDir string `yaml:"base_dir"`

	MaxFileSize     int64 `yaml:"max_file_size"`      // file size on disk
	MaxFileDataSize int64 `yaml:"max_file_data_size"` //uncompressed data size

	Compression bool
	//Delimited enables producing delimited message to text files and length
	//prepended messages to binary files
	FileDelimited bool `yaml:"file_delimited"`

	NonBlocking bool `yaml:"non_blocking"`

	EndOfStreamMark bool

	Encryption EncryptionConfig

	S3     S3Config
	Hadoop HadoopConfig
	Kafka  KafkaConfig

	SQL SQLConfig `yaml:"sql"`
}

PipeConfig holds pipe configuration options

type RowFilter

type RowFilter struct {
	Column    string   `yaml:"column"`
	Values    []string `yaml:"values"`
	Condition string   `yaml:"condition"`
	Operator  string   `yaml:"operator"`
}

RowFilter has the condition, column name & values on which filter will be applied

type S3Config

type S3Config struct {
	Region   string
	Endpoint string
	Bucket   string
	BaseDir  string `yaml:"base_dir"`

	AccessKeyID     string `yaml:"access_key_id"`
	SecretAccessKey string `yaml:"secret_access_key"`
	SessionToken    string `yaml:"session_token"`

	Timeout time.Duration
}

S3Config holds S3 output pipe configuration

type SQLConfig

type SQLConfig struct {
	Type    string
	DSN     string `yaml:"dsn"`
	Service string
	Cluster string
	DB      string
}

SQLConfig holds SQL output pipe configuration

type ScheduleConfig

type ScheduleConfig struct {
	Interval time.Duration //seconds. TODO: Implement proper duration unmarshalling

}

ScheduleConfig holds snapshot schedule parameters

type TableParams

type TableParams struct {
	ClusterConcurrency int   `yaml:"cluster_concurrency"`
	ThrottleTargetMB   int64 `yaml:"throttle_target_mb"`
	ThrottleTargetIOPS int64 `yaml:"throttle_target_iops"`

	Pipe     PipeConfig
	Schedule ScheduleConfig

	RowFilter  RowFilter   `yaml:"row_filter"`
	RowFilters []RowFilter `yaml:"row_filters"` // only used in table params

	ForceIndex string `yaml:"force_index"`

	NoSnapshot bool `yaml:"no_snapshot"`

	// Produce just insert event on update
	NoDeleteOnUpdate bool `yaml:"no_delete_on_update"`
}

TableParams holds per table configuration options

func (*TableParams) CopyForMerge

func (t *TableParams) CopyForMerge() *TableParams

CopyForMerge clear all compound fields in preparation for merge by json.Unmarshal

func (*TableParams) MergeCompound

func (t *TableParams) MergeCompound(r *TableParams)

MergeCompound restores compound fields from original structure if they were empty in merged config