Publish loop:
   send measurements to the aggregator endpoint ':uri',
   every ':frequency' seconds
342887
def publish
      clock = @registry.clock
      until @should_stop
        start = clock.wall_time
        Spectator.logger.info 'Publishing'
        send_metrics_now
        elapsed = clock.wall_time - start
        sleep @frequency - elapsed if elapsed < @frequency
      end
      Spectator.logger.info 'Stopping publishing thread'
    end
945823
function addListeners() {
	if (listenersAdded) return;
	listenersAdded = true;

	if (cluster.isMaster) {
		// Listen for worker responses to requests for local metrics
		cluster.on('message', (worker, message) => {
			if (message.type === GET_METRICS_RES) {
				const request = requests.get(message.requestId);
				message.metrics.forEach(registry => request.responses.push(registry));
				request.pending--;

				if (request.pending === 0) {
					// finalize
					requests.delete(message.requestId);
					clearTimeout(request.errorTimeout);

					if (request.failed) return; // Callback already run with Error.

					const registry = AggregatorRegistry.aggregate(request.responses);
					const promString = registry.metrics();
					request.done(null, promString);
				}
			}
		});
	}
}
970169
public void send(Event event) throws MetricsException {
    HttpPost request = new HttpPost(MetricsUtils.GA_ENDPOINT_URL);
    request.setEntity(MetricsUtils.buildPostBody(event, analyticsId, random));
    try {
      client.execute(request, RESPONSE_HANDLER);
    } catch (IOException e) {
      // All errors in the request execution, including non-2XX HTTP codes,
      // will throw IOException or one of its subclasses.
      throw new MetricsCommunicationException("Problem sending request to server", e);
    }
  }
400437
public int sendAll(Iterable<Observable<Integer>> batches,
                     final int numMetrics, long timeoutMillis) {
    final AtomicBoolean err = new AtomicBoolean(false);
    final AtomicInteger updated = new AtomicInteger(0);
    LOGGER.debug("Got {} ms to send {} metrics", timeoutMillis, numMetrics);
    try {
      final CountDownLatch completed = new CountDownLatch(1);
      final Subscription s = Observable.mergeDelayError(Observable.from(batches))
          .timeout(timeoutMillis, TimeUnit.MILLISECONDS)
          .subscribeOn(Schedulers.immediate())
          .subscribe(updated::addAndGet, exc -> {
            logErr("onError caught", exc, updated.get(), numMetrics);
            err.set(true);
            completed.countDown();
          }, completed::countDown);
      try {
        completed.await(timeoutMillis, TimeUnit.MILLISECONDS);
      } catch (InterruptedException interrupted) {
        err.set(true);
        s.unsubscribe();
        LOGGER.warn("Timed out sending metrics. {}/{} sent", updated.get(), numMetrics);
      }
    } catch (Exception e) {
      err.set(true);
      logErr("Unexpected ", e, updated.get(), numMetrics);
    }

    if (updated.get() < numMetrics && !err.get()) {
      LOGGER.warn("No error caught, but only {}/{} sent.", updated.get(), numMetrics);
    }
    return updated.get();
  }
805337
def publish_delayed_metric(self, name, value, timestamp, raw_value=None,
                               precision=0, metric_type='GAUGE',
                               instance=None):
        
        # Get metric Path
        path = self.get_metric_path(name, instance)

        # Get metric TTL
        ttl = float(self.config['interval']) * float(
            self.config['ttl_multiplier'])

        # Create Metric
        metric = Metric(path, value, raw_value=raw_value, timestamp=timestamp,
                        precision=precision, host=self.get_hostname(),
                        metric_type=metric_type, ttl=ttl)

        # Publish Metric
        self.publish_metric(metric)
374426
public void send(final List<MetricAggregate> aggregates) throws IOException, HttpException {
		
		HttpClient httpClient = new HttpClient(apiConfig);
		
		// retransmit any metrics on the resend queue
		
		resendQueue.drain(httpClient, "/Metrics/SubmitMetricsByID");
		
		// build the json objects
		
		List<JsonMetric> metrics = new ArrayList<JsonMetric>(aggregates.size());
		
		for (MetricAggregate aggregate : aggregates) {
			
			Integer monitorId = monitorService.getMonitorId(aggregate.getIdentity());
			
			if (monitorId != null) {
				JsonMetric.Builder builder = JsonMetric.newBuilder();
				builder.monitorId(monitorId);
				builder.value(Double.valueOf(aggregate.getValue()));
				builder.count(Integer.valueOf(aggregate.getCount()));
				builder.occurredUtc(new Date(aggregate.getOccurredMillis()));
				builder.monitorTypeId(Integer.valueOf(aggregate.getIdentity().getType().getId()));
				builder.clientDeviceId(monitorService.getDeviceId());
				
				metrics.add(builder.build());
			} else {
				LOGGER.info("Unable to determine monitor id for aggregate metric {}", aggregate);
			}
		}
		
		if (metrics.isEmpty()) {
			return;
		}
		
		// post metrics to stackify
		
		byte[] jsonBytes = objectMapper.writer().writeValueAsBytes(metrics);
		
		try {
			httpClient.post("/Metrics/SubmitMetricsByID", jsonBytes);
		} catch (IOException t) {
			LOGGER.info("Queueing metrics for retransmission due to IOException");
			resendQueue.offer(jsonBytes, t);
			throw t;			
		} catch (HttpException t) {
			LOGGER.info("Queueing metrics for retransmission due to HttpException");
			resendQueue.offer(jsonBytes, t);
			throw t;
		}
	}
708773
def send_metrics_to_cloudwatch(self, rule, metric, dimensions):
        

        timestamp = datetime.datetime.utcfromtimestamp(metric.timestamp)

        self.log.debug(
            "CloudWatch: Attempting to publish metric: %s to %s "
            "with value (%s) for dimensions %s @%s",
            rule['name'],
            rule['namespace'],
            str(metric.value),
            str(dimensions),
            str(metric.timestamp)
        )

        try:
            self.connection.put_metric_data(
                str(rule['namespace']),
                str(rule['name']),
                str(metric.value),
                timestamp, str(rule['unit']),
                dimensions)
            self.log.debug(
                "CloudWatch: Successfully published metric: %s to"
                " %s with value (%s) for dimensions %s",
                rule['name'],
                rule['namespace'],
                str(metric.value),
                str(dimensions))
        except AttributeError as e:
            self.log.error(
                "CloudWatch: Failed publishing - %s ", str(e))
        except Exception as e:  # Rough connection re-try logic.
            self.log.error(
                "CloudWatch: Failed publishing - %s\n%s ",
                str(e),
                str(sys.exc_info()[0]))
            self._bind()
1303871
public int flush(final MetricSender sender) throws IOException, HttpException {
		Preconditions.checkNotNull(sender);
				
		// aggregate metrics that were enqueued before the current minute
		
		long currentMinute = (System.currentTimeMillis() / MS_IN_MIN) * MS_IN_MIN;
		
		LOGGER.debug("Flushing metrics < {}", currentMinute);
		LOGGER.debug("Metrics queue size {}", queue.size());
		
		MetricAggregator aggregator = new MetricAggregator(currentMinute, lastValues);
		
		while ((!queue.isEmpty()) && (queue.peek().getOccurredMillis() < currentMinute)) {
			aggregator.add(queue.remove());
		}

		// handle the auto reports
		
		if (lastFlush < currentMinute) {
			aggregator.autoReportZero(autoReportZeroMetrics);
			aggregator.autoReportLast(autoReportLastMetrics);
		}
		
		lastFlush = currentMinute;
		
		// get the aggregates
		
		List<MetricAggregate> aggregates = aggregator.getAggregates();

		// Save the values of gauge and average metrics for the next iteration
		// Gauge last value = aggregate last value
		// Average last value = aggregate last value / aggregate count
		
		if ((aggregates != null) && (!aggregates.isEmpty())) {
			for (MetricAggregate aggregate : aggregates) {
				if (aggregate.getIdentity().getType().equals(MetricMonitorType.GAUGE)) {
					lastValues.put(aggregate.getIdentity(), aggregate.getValue());
				} else if (aggregate.getIdentity().getType().equals(MetricMonitorType.AVERAGE)) {
					lastValues.put(aggregate.getIdentity(), aggregate.getValue() / aggregate.getCount());
				}
			}
		}
		
		// send the aggregates to Stackify
		
		int numSent = 0;
		
		if ((aggregates != null) && (!aggregates.isEmpty())) {
			LOGGER.debug("Sending aggregate metrics: {}", aggregates);
			sender.send(aggregates);
			numSent = aggregates.size();
		}
		
		return numSent;
	}
330498
def publish(self, topic, payload = None, qos = 0, retain = False):
        """"""
        #print "PUBLISHING (",topic,"): ", payload
        payloadlen = len(payload)
        if topic is None or qos < 0 or qos > 2:
            print "PUBLISH:err inval"
            return NC.ERR_INVAL
        
        #payloadlen <= 250MB
        if payloadlen > (250 * 1024 * 1204):
            self.logger.error("PUBLISH:err payload len:%d", payloadlen)
            return NC.ERR_PAYLOAD_SIZE
        
        #wildcard check : TODO
        mid = self.mid_generate()
        
        if qos in (0,1,2):
            return self.send_publish(mid, topic, payload, qos, retain, False)

        else:
            self.logger.error("Unsupport QoS= %d", qos)
    
        return NC.ERR_NOT_SUPPORTED
833908
@Scheduled(fixedRateString = "${hystrix.stream.queue.sendRate:${hystrix.stream.queue.send-rate:500}}")
	public void sendMetrics() {
		ArrayList<String> metrics = new ArrayList<>();
		this.jsonMetrics.drainTo(metrics);

		if (!metrics.isEmpty()) {
			if (log.isTraceEnabled()) {
				log.trace("sending stream metrics size: " + metrics.size());
			}
			for (String json : metrics) {
				// TODO: batch all metrics to one message
				try {
					// TODO: remove the explicit content type when s-c-stream can handle
					// that for us
					this.outboundChannel.send(MessageBuilder.withPayload(json)
							.setHeader(MessageHeaders.CONTENT_TYPE,
									this.properties.getContentType())
							.build());
				}
				catch (Exception ex) {
					if (log.isTraceEnabled()) {
						log.trace("failed sending stream metrics: " + ex.getMessage());
					}
				}
			}
		}
	}
710749
def _collect_metrics_async(self, instance, query_specs):
        
        # ## <TEST-INSTRUMENTATION>
        t = Timer()
        # ## </TEST-INSTRUMENTATION>
        i_key = self._instance_key(instance)
        server_instance = self._get_server_instance(instance)
        perfManager = server_instance.content.perfManager
        custom_tags = instance.get('tags', [])
        results = perfManager.QueryPerf(query_specs)
        if results:
            for mor_perfs in results:
                mor_name = str(mor_perfs.entity)
                try:
                    mor = self.mor_cache.get_mor(i_key, mor_name)
                except MorNotFoundError:
                    self.log.error(
                        "Trying to get metrics from object %s deleted from the cache, skipping. "
                        "Consider increasing the parameter `clean_morlist_interval` to avoid that",
                        mor_name,
                    )
                    continue

                for result in mor_perfs.value:
                    counter_id = result.id.counterId
                    if not self.metadata_cache.contains(i_key, counter_id):
                        self.log.debug(
                            "Skipping value for counter {}, because there is no metadata about it".format(
                                ensure_unicode(counter_id)
                            )
                        )
                        continue

                    # Metric types are absolute, delta, and rate
                    metric_name = self.metadata_cache.get_metadata(i_key, result.id.counterId).get('name')

                    if self.in_compatibility_mode(instance):
                        if metric_name not in ALL_METRICS:
                            self.log.debug("Skipping unknown `{}` metric.".format(ensure_unicode(metric_name)))
                            continue

                    if not result.value:
                        self.log.debug(
                            "Skipping `{}` metric because the value is empty".format(ensure_unicode(metric_name))
                        )
                        continue

                    instance_name = result.id.instance or "none"
                    value = self._transform_value(instance, result.id.counterId, result.value[0])

                    tags = ['instance:{}'.format(ensure_unicode(instance_name))]
                    if not mor['hostname']:  # no host tags available
                        tags.extend(mor['tags'])

                    # vsphere "rates" should be submitted as gauges (rate is
                    # precomputed).
                    self.gauge(
                        "vsphere.{}".format(ensure_unicode(metric_name)),
                        value,
                        hostname=mor['hostname'],
                        tags=tags + custom_tags,
                    )

        # ## <TEST-INSTRUMENTATION>
        self.histogram('datadog.agent.vsphere.metric_colection.time', t.total(), tags=custom_tags)
21799
def _publish_transformed(self, data, base_prefix):
        
        self._publish_dict_with_prefix(data.get('opcounters', {}),
                                       base_prefix + ['opcounters_per_sec'],
                                       self.publish_counter)
        self._publish_dict_with_prefix(data.get('opcountersRepl', {}),
                                       base_prefix +
                                       ['opcountersRepl_per_sec'],
                                       self.publish_counter)
        self._publish_dict_with_prefix(data.get('network', {}),
                                       base_prefix + ['network_per_sec'],
                                       self.publish_counter)
        self._publish_metrics(base_prefix + ['extra_info_per_sec'],
                              'page_faults',
                              data.get('extra_info', {}),
                              self.publish_counter)

        def get_dotted_value(data, key_name):
            key_name = key_name.split('.')
            for i in key_name:
                data = data.get(i, {})
                if not data:
                    return 0
            return data

        def compute_interval(data, total_name):
            current_total = get_dotted_value(data, total_name)
            total_key = '.'.join(base_prefix + [total_name])
            last_total = self.__totals.get(total_key, current_total)
            interval = current_total - last_total
            self.__totals[total_key] = current_total
            return interval

        def publish_percent(value_name, total_name, data):
            value = float(get_dotted_value(data, value_name) * 100)
            interval = compute_interval(data, total_name)
            key = '.'.join(base_prefix + ['percent', value_name])
            self.publish_counter(key, value, time_delta=bool(interval),
                                 interval=interval)

        publish_percent('globalLock.lockTime', 'globalLock.totalTime', data)

        locks = data.get('locks')
        if locks:
            if '.' in locks:
                locks['_global_'] = locks['.']
                del (locks['.'])
            key_prefix = '.'.join(base_prefix + ['percent'])
            db_name_filter = re.compile(self.config['databases'])
            interval = compute_interval(data, 'uptimeMillis')
            for db_name in locks:
                if not db_name_filter.search(db_name):
                    continue
                r = get_dotted_value(
                    locks,
                    '%s.timeLockedMicros.r' % db_name)
                R = get_dotted_value(
                    locks,
                    '.%s.timeLockedMicros.R' % db_name)
                value = float(r + R) / 10
                if value:
                    self.publish_counter(
                        key_prefix + '.locks.%s.read' % db_name,
                        value, time_delta=bool(interval),
                        interval=interval)
                w = get_dotted_value(
                    locks,
                    '%s.timeLockedMicros.w' % db_name)
                W = get_dotted_value(
                    locks,
                    '%s.timeLockedMicros.W' % db_name)
                value = float(w + W) / 10
                if value:
                    self.publish_counter(
                        key_prefix + '.locks.%s.write' % db_name,
                        value, time_delta=bool(interval), interval=interval)
864961
def process(self, metric):
        
        path = metric.getCollectorPath()
        path += '.'
        path += metric.getMetricPath()
        if self.config['apply_metric_prefix']:
            path = metric.getPathPrefix() + '.' + path

        if self.include_reg.match(path):
            if metric.metric_type == 'GAUGE':
                m_type = 'gauge'
            else:
                m_type = 'counter'
            self.queue.add(path,                # name
                           float(metric.value),  # value
                           type=m_type,
                           source=metric.host,
                           measure_time=metric.timestamp)
            self.current_n_measurements += 1
        else:
            self.log.debug("LibratoHandler: Skip %s, no include_filters match",
                           path)

        if (self.current_n_measurements >= self.queue_max_size or
                time.time() >= self.queue_max_timestamp):
            self.log.debug("LibratoHandler: Sending batch size: %d",
                           self.current_n_measurements)
            self._send()
339430
def scrape_metrics(self, endpoint):
        
        response = self.poll(endpoint)
        try:
            # no dry run if no label joins
            if not self.label_joins:
                self._dry_run = False
            elif not self._watched_labels:
                # build the _watched_labels set
                for val in itervalues(self.label_joins):
                    self._watched_labels.add(val['label_to_match'])

            for metric in self.parse_metric_family(response):
                yield metric

            # Set dry run off
            self._dry_run = False
            # Garbage collect unused mapping and reset active labels
            for metric, mapping in list(iteritems(self._label_mapping)):
                for key in list(mapping):
                    if key not in self._active_label_mapping[metric]:
                        del self._label_mapping[metric][key]
            self._active_label_mapping = {}
        finally:
            response.close()
943280
def _publish_metrics(self, prev_keys, key, data, publishfn=None):
        """"""
        if key not in data:
            return
        value = data[key]
        keys = prev_keys + [key]
        if not publishfn:
            publishfn = self.publish
        if isinstance(value, dict):
            for new_key in value:
                self._publish_metrics(keys, new_key, value)
        elif isinstance(value, int) or isinstance(value, float):
            publishfn('.'.join(keys), value)
        elif isinstance(value, long):
            publishfn('.'.join(keys), float(value))
94814
function(obj, methodName, methodArgs, probeData) {
        // Start metrics
        that.metricsProbeStart(probeData);
        that.requestProbeStart(probeData);
        // End metrics
        aspect.aroundCallback(
          methodArgs,
          probeData,
          function(target, args, probeData) {
            // Get HTTP request method from options
            var ri = getRequestItems(methodArgs[0]);
            that.metricsProbeEnd(probeData, ri.requestMethod, ri.urlRequested, args[0], ri.headers);
            that.requestProbeEnd(probeData, ri.requestMethod, ri.urlRequested, args[0], ri.headers);
          },
          function(target, args, probeData, ret) {
            // Don't need to do anything after the callback
            return ret;
          }
        );
      }
132797
def emit(_):
    """"""
    if not initialized:
        raise NotInitialized

    view = {
        'version': __version__,
        'counters': {},
        'gauges': {},
        'histograms': {},
        'meters': {},
        'timers': {},
    }

    for (ty, module, name), metric in six.iteritems(all_metrics):
        view[ty]['%s.%s' % (module, name)] = metric.view()

    marshalled_view = marshal.dumps(view)
    if len(marshalled_view) > MAX_MARSHALLED_VIEW_SIZE:
        log.warn(
            'Marshalled length too large, got %d, max %d. '
            'Try recording fewer metrics or increasing '
            'MAX_MARSHALLED_VIEW_SIZE'
            % (len(marshalled_view), MAX_MARSHALLED_VIEW_SIZE))
        return
    marshalled_metrics_mmap.seek(0)
    try:
        # Reading and writing to/from an mmap'ed buffer is not guaranteed
        # to be atomic, so we must serialize access to it.
        uwsgi.lock()
        marshalled_metrics_mmap.write(marshalled_view)
    finally:
        uwsgi.unlock()