Golang fullerite-collector.Collector类(方法)实例源码

func reportCollector(collector collector.Collector) {
	log.Warn(fmt.Sprintf("%s collector took too long to run, reporting incident!", collector.Name()))
	metric := metric.New("fullerite.collection_time_exceeded")
	metric.Value = 1
	metric.AddDimension("interval", fmt.Sprintf("%d", collector.Interval()))
	collector.Channel() <- metric
}

func runCollector(collector collector.Collector) {
	log.Info("Running ", collector)

	ticker := time.NewTicker(time.Duration(collector.Interval()) * time.Second)
	collect := ticker.C

	staggerValue := 1
	collectionDeadline := time.Duration(collector.Interval() + staggerValue)

	for {
		select {
		case <-collect:
			if collector.CollectorType() == "listener" {
				collector.Collect()
			} else {
				countdownTimer := time.AfterFunc(collectionDeadline*time.Second, func() {
					reportCollector(collector)
				})
				collector.Collect()
				countdownTimer.Stop()
			}
		}
	}
	ticker.Stop()
}

func runCollector(collector collector.Collector) {
	log.Info("Running ", collector)
	for {
		collector.Collect()
		time.Sleep(time.Duration(collector.Interval()) * time.Second)
	}
}

func runCollector(collector collector.Collector) {
	for {
		log.Println("Collecting from", collector)
		collector.Collect()
		time.Sleep(time.Duration(collector.Interval()) * time.Second)
	}
}

func readFromCollector(collector collector.Collector, metrics chan metric.Metric) {
	for metric := range collector.Channel() {
		if _, exists := metric.GetDimensionValue("collector"); !exists {
			metric.AddDimension("collector", collector.Name())
		}
		metrics <- metric
	}
}

func readFromCollector(collector collector.Collector,
	handlers []handler.Handler,
	collectorStatChans ...chan<- metric.CollectorEmission) {
	// In case of Diamond collectors, metric from multiple collectors are read
	// from Single channel (owned by Go Diamond Collector) and hence we use a map
	// for keeping track of metrics from each individual collector
	emissionCounter := map[string]uint64{}
	lastEmission := time.Now()
	statDuration := time.Duration(collector.Interval()) * time.Second
	for m := range collector.Channel() {
		var exists bool
		c := collector.CanonicalName()
		if _, exists = m.GetDimensionValue("collector"); !exists {
			m.AddDimension("collector", collector.Name())
		}
		// We allow external collectors to provide us their collector's CanonicalName
		// by sending it as a metric dimension. For example in the case of Diamond the
		// individual python collectors can send their names this way.
		if val, ok := m.GetDimensionValue("collectorCanonicalName"); ok {
			c = val
			m.RemoveDimension("collectorCanonicalName")
		}
		emissionCounter[c]++
		// collectorStatChans is an optional parameter. In case of ad-hoc collector
		// this parameter is not supplied at all. Using variadic arguments is pretty much
		// only way of doing this in go.
		if len(collectorStatChans) > 0 {
			collectorStatChan := collectorStatChans[0]
			currentTime := time.Now()
			if currentTime.After(lastEmission.Add(statDuration)) {
				emitCollectorStats(emissionCounter, collectorStatChan)
				lastEmission = time.Now()
			}
		}

		for i := range handlers {
			if _, exists := handlers[i].CollectorChannels()[c]; exists {
				handlers[i].CollectorChannels()[c] <- m
			}
		}
	}
	// Closing the stat channel after collector loop finishes
	for _, statChannel := range collectorStatChans {
		close(statChannel)
	}
}

func readFromCollector(collector collector.Collector, handlers []handler.Handler) {
	for metric := range collector.Channel() {
		var exists bool
		c := collector.CanonicalName()
		if _, exists = metric.GetDimensionValue("collector"); !exists {
			metric.AddDimension("collector", collector.Name())
		}
		// We allow external collectors to provide us their collector's CanonicalName
		// by sending it as a metric dimension. For example in the case of Diamond the
		// individual python collectors can send their names this way.
		if val, ok := metric.GetDimensionValue("collectorCanonicalName"); ok {
			c = val
			metric.RemoveDimension("collectorCanonicalName")
		}

		for i := range handlers {
			if _, exists := handlers[i].CollectorChannels()[c]; exists {
				handlers[i].CollectorChannels()[c] <- metric
			}
		}
	}
}

func readFromCollector(collector collector.Collector, metrics chan metric.Metric) {
	for metric := range collector.Channel() {
		metric.AddDimension("collector", collector.Name())
		metrics <- metric
	}
}

func readFromCollector(collector collector.Collector, metrics chan metric.Metric) {
	for metric := range collector.Channel() {
		metrics <- metric
	}
}

func readCollectorConfig(collector collector.Collector) {
	// TODO: actually read from configuration file.
	collector.SetInterval(10)
}