Queues a list of metrics for a specified set of object IDs asynchronously
543770
def queue_async_stats_job(klass, account, ids, metric_groups, **kwargs):
        
        params = klass._standard_params(ids, metric_groups, **kwargs)

        params['platform'] = kwargs.get('platform', None)
        params['country'] = kwargs.get('country', None)
        params['segmentation_type'] = kwargs.get('segmentation_type', None)

        resource = klass.RESOURCE_ASYNC.format(account_id=account.id)
        response = Request(account.client, 'post', resource, params=params).perform()
        return response.body['data']
691541
def _spark_executor_metrics(self, instance, running_apps, addl_tags, requests_config):
        
        for app_id, (app_name, tracking_url) in iteritems(running_apps):

            base_url = self._get_request_url(instance, tracking_url)
            response = self._rest_request_to_json(
                base_url, SPARK_APPS_PATH, SPARK_SERVICE_CHECK, requests_config, addl_tags, app_id, 'executors'
            )

            tags = ['app_name:%s' % str(app_name)]
            tags.extend(addl_tags)

            for executor in response:
                if executor.get('id') == 'driver':
                    self._set_metrics_from_json(tags, executor, SPARK_DRIVER_METRICS)
                else:
                    self._set_metrics_from_json(tags, executor, SPARK_EXECUTOR_METRICS)

            if len(response):
                self._set_metric('spark.executor.count', COUNT, len(response), tags)
771140
def StreamMetrics(self, request_iterator, context):
        """"""
        LOG.debug("StreamMetrics called")

        # set up arguments
        collect_args = (next(request_iterator))
        max_metrics_buffer = 0
        max_collect_duration = 0
        cfg = Metric(pb=collect_args.Metrics_Arg.metrics[0])
        try:
            max_metrics_buffer = int(cfg.config["max-metrics-buffer"])
        except Exception as ex:
            LOG.debug("Unable to get schedule parameters: {}".format(ex))
        try:
            max_collect_duration = int(cfg.config["max-collect-duration"])
        except Exception as ex:
            LOG.debug("Unable to get schedule parameters: {}".format(ex))
        if max_metrics_buffer > 0:
            self.max_metrics_buffer = max_metrics_buffer
        if max_collect_duration > 0:
            self.max_collect_duration = max_collect_duration

        # start collection thread
        thread = threading.Thread(target=self._stream_wrapper, args=(collect_args,),)
        thread.daemon = True
        thread.start()

        # stream metrics
        metrics = []
        metrics_to_stream = []
        stream_timeout = self.max_collect_duration
        while context.is_active():
            try:
                # wait for metrics until timeout is reached
                t_start = time.time()
                metrics = self.metrics_queue.get(block=True, timeout=stream_timeout)
                elapsed = round(time.time() - t_start)
                stream_timeout -= elapsed
            except queue.Empty:
                LOG.debug("Max collect duration exceeded. Streaming {} metrics".format(len(metrics_to_stream)))
                metrics_col = CollectReply(Metrics_Reply=MetricsReply(metrics=[m.pb for m in metrics_to_stream]))
                metrics_to_stream = []
                stream_timeout = self.max_collect_duration
                yield metrics_col
            else:
                for metric in metrics:
                    metrics_to_stream.append(metric)
                    if len(metrics_to_stream) == self.max_metrics_buffer:
                        LOG.debug("Max metrics buffer reached. Streaming {} metrics".format(len(metrics_to_stream)))
                        metrics_col = CollectReply(
                            Metrics_Reply=MetricsReply(metrics=[m.pb for m in metrics_to_stream]))
                        metrics_to_stream = []
                        stream_timeout = self.max_collect_duration
                        yield metrics_col
                # stream metrics if max_metrics_buffer is 0 or enough metrics has been collected
                if self.max_metrics_buffer == 0:
                    LOG.debug("Max metrics buffer set to 0. Streaming {} metrics".format(len(metrics_to_stream)))
                    metrics_col = CollectReply(Metrics_Reply=MetricsReply(metrics=[m.pb for m in metrics_to_stream]))
                    metrics_to_stream = []
                    stream_timeout = self.max_collect_duration
                    yield metrics_col

        # sent notification if stream has been stopped
        self.done_queue.put(True)
473156
def _view_metrics(self, queue_type, queue_id):
        """"""
        response = {
            'status': 'failure'
        }
        request_data = {}
        if queue_type:
            request_data['queue_type'] = queue_type
        if queue_id:
            request_data['queue_id'] = queue_id

        try:
            response = self.sq.metrics(**request_data)
        except Exception, e:
            response['message'] = e.message
            return jsonify(**response), 400

        return jsonify(**response)
427500
def _run_queued_callbacks(self):
        '''  '''
        for callback in self._queued_callbacks:
            try:
                callback()
            except Exception as ex:
                logger.error("Exception: %s" % str(ex))
261729
static Deferred<byte[]> rowKeyTemplateAsync(final TSDB tsdb,
      final String metric, final Map<String, String> tags) {
    final short metric_width = tsdb.metrics.width();
    final short tag_name_width = tsdb.tag_names.width();
    final short tag_value_width = tsdb.tag_values.width();
    final short num_tags = (short) tags.size();

    int row_size = (Const.SALT_WIDTH() + metric_width + Const.TIMESTAMP_BYTES 
        + tag_name_width * num_tags + tag_value_width * num_tags);
    final byte[] row = new byte[row_size];

    // Lookup or create the metric ID.
    final Deferred<byte[]> metric_id;
    if (tsdb.config.auto_metric()) {
      metric_id = tsdb.metrics.getOrCreateIdAsync(metric, metric, tags);
    } else {
      metric_id = tsdb.metrics.getIdAsync(metric);
    }

    // Copy the metric ID at the beginning of the row key.
    class CopyMetricInRowKeyCB implements Callback<byte[], byte[]> {
      public byte[] call(final byte[] metricid) {
        copyInRowKey(row, (short) Const.SALT_WIDTH(), metricid);
        return row;
      }
    }

    // Copy the tag IDs in the row key.
    class CopyTagsInRowKeyCB implements
        Callback<Deferred<byte[]>, ArrayList<byte[]>> {
      public Deferred<byte[]> call(final ArrayList<byte[]> tags) {
        short pos = (short) (Const.SALT_WIDTH() + metric_width);
        pos += Const.TIMESTAMP_BYTES;
        for (final byte[] tag : tags) {
          copyInRowKey(row, pos, tag);
          pos += tag.length;
        }
        // Once we've resolved all the tags, schedule the copy of the metric
        // ID and return the row key we produced.
        return metric_id.addCallback(new CopyMetricInRowKeyCB());
      }
    }

    // Kick off the resolution of all tags.
    return Tags.resolveOrCreateAllAsync(tsdb, metric, tags)
        .addCallbackDeferring(new CopyTagsInRowKeyCB());
  }
961977
protected synchronized void compareReply(List<Long> processIds) {
        Object[] replyIds = replyProcessIds.toArray();
        for (Object replyId : replyIds) {
            if (processIds.contains((Long) replyId) == false) { // 判断reply id是否在当前processId列表中
                // 因为存在并发问题，如在执行Listener事件的同时，可能触发了process的创建，这时新建的processId会进入到reply队列中
                // 此时接受到的processIds变量为上一个版本的内容，所以会删除新建的process，导致整个通道被挂住
                if (CollectionUtils.isEmpty(processIds) == false) {
                    Long processId = processIds.get(0);
                    if (processId > (Long) replyId) { // 如果当前最小的processId都大于replyId, processId都是递增创建的
                        processTermined((Long) replyId); // 调用一下删除操作
                    }
                }
            }
        }
    }
937532
def bulk_write(self, metrics):
        
        actions = []
        index = self.get_index()
        for metric in metrics:
            actions.append({'index': {'_index': index, '_type': self.doc_type}})
            actions.append(metric)
        try:
            self.client.bulk(actions)
        except TransportError as exc:
            logger.warning('bulk_write metrics %r failure %r', metrics, exc)
600121
private void queryMetrics(final MetricQueryServiceGateway queryServiceGateway) {
		LOG.debug("Query metrics for {}.", queryServiceGateway.getAddress());

		queryServiceGateway
			.queryMetrics(timeout)
			.whenCompleteAsync(
				(MetricDumpSerialization.MetricSerializationResult result, Throwable t) -> {
					if (t != null) {
						LOG.debug("Fetching metrics failed.", t);
					} else {
						metrics.addAll(deserializer.deserialize(result));
					}
				},
				executor);
	}
217515
private Response serveOneOrAll(List<water.ModelMetrics> list) {
    JsonArray metricsArray = new JsonArray();

    for (water.ModelMetrics metrics : list) {
      JsonObject metricsJson = metrics.toJSON();
      metricsArray.add(metricsJson);
    }

    JsonObject result = new JsonObject();
    result.add("metrics", metricsArray);
    return Response.done(result);
  }
210189
def get_waiting_keys(cls, names):
        
        return [queue.waiting.key for queue in cls.get_all_by_priority(names)]
793417
def collect(self):
        
        performance_indicators = {
            'ns': 'network_send',
            'nr': 'network_receive',
            'dw': 'disk_write',
            'dr': 'disk_read',
            'al': 'activity_log',
            'bm': 'bit_map',
            'lo': 'local_count',
            'pe': 'pending',
            'ua': 'unacknowledged',
            'ap': 'application_pending',
            'ep': 'epochs',
            'wo': 'write_order',
            'oos': 'out_of_sync',
            'cs': 'connection_state',
            'ro': 'roles',
            'ds': 'disk_states'
        }

        results = dict()
        try:
            statusfile = open('/proc/drbd', 'r')
            current_resource = ''
            for line in statusfile:
                if re.search('version', line) is None:
                    if re.search(r' \d: cs', line):
                        matches = re.match(r' (\d): (cs:\w+) (ro:\w+/\w+) '
                                           '(ds:\w+/\w+) (\w{1}) .*', line)
                        current_resource = matches.group(1)
                        results[current_resource] = dict()
                    elif re.search(r'\sns:', line):
                        metrics = line.strip().split(" ")
                        for metric in metrics:
                            item, value = metric.split(":")
                            results[current_resource][
                                performance_indicators[item]] = value

                else:
                    continue
            statusfile.close()
        except IOError as errormsg:
            self.log.error("Can't read DRBD status file: {}".format(errormsg))
            return

        for resource in results.keys():
            for metric_name, metric_value in results[resource].items():
                if metric_value.isdigit():
                    self.publish(resource + "." + metric_name, metric_value)
                else:
                    continue
1099556
@Override
  public void getMetrics(MetricsCollector collector, boolean all) {
    MetricsRecordBuilder metricsRecordBuilder = collector.addRecord("sparkShuffleService");

    for (Map.Entry<String, Metric> entry : metricSet.getMetrics().entrySet()) {
      collectMetric(metricsRecordBuilder, entry.getKey(), entry.getValue());
    }
  }
181748
def run(self):
        

        services = Service.objects.all()
        for service in services:
            service_metric_sync.apply_async(
                kwargs={"service_id": str(service.id)})

            key = "services.downtime.%s.sum" % (
                utils.normalise_string(service.name))
            check = WidgetData.objects.filter(service=None, key=key)
            if not check.exists():
                WidgetData.objects.create(
                    key=key,
                    title="TEMP - Pending update"
                )

        return "Queued <%s> Service(s) for Metric Sync" % services.count()
324485
def _get_batch_requests(self, timeout=None):
        """"""
        reqs = []
        s = time()
        while len(reqs) < self.batch_size and (time() - s) < timeout:
            try:
                req = self.queue.get(block=False)
                self.queue.task_done()

                reqs.append(req)
            except Empty as ex:
                if self.stop_flag:
                    break
                else:
                    sleep(0.1)

        if not reqs:
            raise Empty
        elif len(reqs) <= 1:
            return reqs[0]
        else:
            logitems = []
            req = reqs[0]
            for req in reqs:
                logitems.extend(req.get_log_items())

            ret = PutLogsRequest(self.project, self.log_store, req.topic, logitems=logitems)
            ret.__record__ = req.__record__

            return ret
1124513
def compute_file_metrics(processors, language, key, token_list):
    """"""
    # multiply iterator
    tli = itertools.tee(token_list, len(processors))
    metrics = OrderedDict()

    # reset all processors
    for p in processors:
        p.reset()

    # process all tokens
    for p, tl in zip(processors, tli):
        p.process_file(language, key, tl)

    # collect metrics from all processors
    for p in processors:
        metrics.update(p.metrics)

    return metrics
199662
def _collect_monitor_metrics(self, conn, tags):
        
        for entry in conn.entries:
            # Get metrics from monitor backend
            dn = entry.entry_dn.lower()
            if dn.endswith(self.CONNECTIONS_METRICS_DN):
                self._handle_connections_entry(entry, tags)
            elif dn.endswith(self.OPERATIONS_METRICS_DN):
                self._handle_operations_entry(entry, tags)
            elif dn.endswith(self.STATISTICS_METRICS_DN):
                self._handle_statistics_entry(entry, tags)
            elif dn.endswith(self.THREADS_METRICS_DN):
                self._handle_threads_entry(entry, tags)
            elif dn.endswith(self.TIME_METRICS_DN):
                self._handle_time_entry(entry, tags)
            elif dn.endswith(self.WAITERS_METRICS_DN):
                self._handle_waiters_entry(entry, tags)