cmds

Documentation

Index

• Constants
• Variables
• func ActionBaritoConsumerService(c *cli.Context) (err error)
• func ActionBaritoProducerService(c *cli.Context) (err error)
• type ApplicationGroup
• type MetricMarketCallback
  • func NewMetricMarketCallback(url string) MetricMarketCallback
• type MetricPayload

Constants

const (
	EnvKafkaBrokers       = "BARITO_KAFKA_BROKERS"
	EnvKafkaGroupID       = "BARITO_KAFKA_GROUP_ID"
	EnvKafkaTopicSuffix   = "BARITO_KAFKA_TOPIC_SUFFIX"
	EnvKafkaMaxRetry      = "BARITO_KAFKA_MAX_RETRY"
	EnvKafkaRetryInterval = "BARITO_KAFKA_RETRY_INTERVAL"

	EnvElasticsearchUrls = "BARITO_ELASTICSEARCH_URLS"
	EnvEsIndexMethod     = "BARITO_ELASTICSEARCH_INDEX_METHOD"
	EnvEsBulkSize        = "BARITO_ELASTICSEARCH_BULK_SIZE"
	EnvEsFlushIntervalMs = "BARITO_ELASTICSEARCH_FLUSH_INTERVAL_MS"

	EnvGrpcMaxRecvMsgSize = "BARITO_GRPC_MAX_RECV_MSG_SIZE"

	EnvPushMetricUrl      = "BARITO_PUSH_METRIC_URL"
	EnvPushMetricInterval = "BARITO_PUSH_METRIC_INTERVAL"

	EnvServeRestApi                   = "BARITO_PRODUCER_REST_API" // TODO: rename to better name
	EnvProducerAddressGrpc            = "BARITO_PRODUCER_GRPC"     // TODO: rename to better name
	EnvProducerAddressRest            = "BARITO_PRODUCER_REST"     // TODO: rename to better name
	EnvProducerMaxRetry               = "BARITO_PRODUCER_MAX_RETRY"
	EnvProducerMaxTPS                 = "BARITO_PRODUCER_MAX_TPS"
	EnvProducerRateLimitResetInterval = "BARITO_PRODUCER_RATE_LIMIT_RESET_INTERVAL"

	EnvConsulUrl               = "BARITO_CONSUL_URL"
	EnvConsulKafkaName         = "BARITO_CONSUL_KAFKA_NAME"
	EnvConsulElasticsearchName = "BARITO_CONSUL_ELASTICSEARCH_NAME"

	EnvNewTopicEventName                    = "BARITO_NEW_TOPIC_EVENT"
	EnvConsumerElasticsearchRetrierInterval = "BARITO_CONSUMER_ELASTICSEARCH_RETRIER_INTERVAL"
	EnvConsumerElasticsearchRetrierMaxRetry = "BARITO_CONSUMER_ELASTICSEARCH_RETRIER_MAX_RETRY"
	EnvConsumerRebalancingStrategy          = "BARITO_CONSUMER_REBALANCING_STRATEGY"

	EnvPrintTPS = "BARITO_PRINT_TPS"

	EnvElasticUsername = "ELASTIC_USERNAME"
	EnvElasticPassword = "ELASTIC_PASSWORD"
)

Variables

var (
	DefaultConsulKafkaName         = "kafka"
	DefaultConsulElasticsearchName = "elasticsearch"

	DefaultKafkaBrokers       = []string{"localhost:9092"}
	DefaultKafkaTopicSuffix   = "_logs"
	DefaultKafkaGroupID       = "barito-group"
	DefaultKafkaMaxRetry      = 0
	DefaultKafkaRetryInterval = 10

	DefaultElasticsearchUrls = []string{"http://localhost:9200"}

	DefaultGrpcMaxRecvMsgSize = 20 * 1000 * 1000

	DefaultPushMetricUrl      = ""
	DefaultPushMetricInterval = "30s"

	DefaultServeRestApi                   = "true"
	DefaultProducerAddressGrpc            = ":8082"
	DefaultProducerAddressRest            = ":8080"
	DefaultProducerMaxRetry               = 10
	DefaultProducerMaxTPS                 = 100
	DefaultProducerRateLimitResetInterval = 10

	DefaultNewTopicEventName            = "new_topic_events"
	DefaultElasticsearchRetrierInterval = "30s"
	DefaultElasticsearchRetrierMaxRetry = 10
	DefaultConsumerRebalancingStrategy  = "RoundRobin"
	DefaultEsIndexMethod                = "BulkProcessor"
	DefaultEsBulkSize                   = 100
	DefaultEsFlushIntervalMs            = 500

	DefaultPrintTPS = "false"

	DefaultElasticUsername = ""
	DefaultElasticPassword = ""
)

Functions

func ActionBaritoConsumerService

func ActionBaritoConsumerService(c *cli.Context) (err error)

func ActionBaritoProducerService

func ActionBaritoProducerService(c *cli.Context) (err error)

Types

type ApplicationGroup

type ApplicationGroup struct {
	AppSecret string `json:"token"`
	LogCount  int64  `json:"new_log_count"`
}

type MetricMarketCallback

type MetricMarketCallback interface {
	OnCallback(instr instru.Instrumentation) error
}

TODO: move to flow package

func NewMetricMarketCallback

func NewMetricMarketCallback(url string) MetricMarketCallback

type MetricPayload

type MetricPayload struct {
	ApplicationGroups []ApplicationGroup `json:"application_groups"`
}