collector

Documentation

Index

• Variables
• func ConvertMonMetric(metric types.Metric, metricVal TelegrafMetric) (map[string]interface{}, error)
• type Aggregator
  • func (a *Aggregator) AggregateMetric(kafkaConn *kafka.Consumer, topics []string) ([]string, error)
  • func (a *Aggregator) CalculateMetric(responseMap map[string]map[string]map[string][]float64, ...) (map[string]interface{}, error)
• type MetricCollector
  • func NewMetricCollector(aggregateType types.AggregateType, createOrder int) (MetricCollector, error)
  • func (mc *MetricCollector) Collector(wg *sync.WaitGroup) error
• type TelegrafMetric

Variables

var KafkaConfig *kafka.ConfigMap

Functions

func ConvertMonMetric

func ConvertMonMetric(metric types.Metric, metricVal TelegrafMetric) (map[string]interface{}, error)

Types

type Aggregator

type Aggregator struct {
	AggregateType types.AggregateType
}

func (*Aggregator) AggregateMetric

func (a *Aggregator) AggregateMetric(kafkaConn *kafka.Consumer, topics []string) ([]string, error)

func (*Aggregator) CalculateMetric

func (a *Aggregator) CalculateMetric(responseMap map[string]map[string]map[string][]float64, tagMap map[string]map[string]string, aggregateType string) (map[string]interface{}, error)

type MetricCollector

type MetricCollector struct {
	CreateOrder       int
	ConsumerKafkaConn *kafka.Consumer
	Aggregator        Aggregator
	Ch                chan []string
}

func NewMetricCollector

func NewMetricCollector(aggregateType types.AggregateType, createOrder int) (MetricCollector, error)

NewMetricCollector

• Go-routine 기반 collector 입니다.
• go channel 및 Kafka 기반으로 topic 분배받기 & topic 구독 및 가져오기를 수행합니다.

func (*MetricCollector) Collector

func (mc *MetricCollector) Collector(wg *sync.WaitGroup) error

Collector

• 콜렉터 매니저로부터 "close" 채널 값을 받으면 종료합니다. (고루틴 채널 중지 => 삭제)
• 콜렉터 매니저로부터 topic 리스트 값을 받으면 kafka 에 해당 topic 을 기준으로 데이터를 가져옵니다.
• kafka 에 요청한 topic 리스트 들 중 데이터가 3회 이상 넘어오지 않는 topic 의 경우 < 스케줄러가 활용하는 topic Queue > 에 삭제할 topic 으로 등록합니다.

type TelegrafMetric

type TelegrafMetric struct {
	Name      string                 `json:"name"`
	Tags      map[string]interface{} `json:"tags"`
	Fields    map[string]interface{} `json:"fields"`
	Timestamp int64                  `json:"timestamp"`
	TagInfo   map[string]interface{} `json:"tagInfo"`
}