Create and register metric,
        find subscribers for this metric (using meta as filter) and subscribe

        Return:
            metric (available_metrics[0]): one of Metric
324307
def new_tank_metric(self, _type, name, hostname=None, group=None, source=None, *kw):
        
        if not _type:
            raise ValueError('Metric type should be defined.')

        if _type in available_metrics:
            metric_obj = available_metrics[_type](None, self.routing_queue)  # create metric object
            metric_info = {'type': _type,
                           'source': source,
                           'name': name,
                           'hostname': hostname,
                           'group': group}
            if kw is not None:
                metric_info.update(kw)
            metric_meta = pd.DataFrame({metric_obj.local_id: metric_info}).T  # create metric meta
            self.metrics_meta = self.metrics_meta.append(metric_meta)  # register metric meta
            self.metrics[metric_obj.local_id] = metric_obj  # register metric object

            # find subscribers for this metric
            this_metric_subscribers = self.__reversed_filter(self.subscribers, metric_info)
            if this_metric_subscribers.empty:
                logger.debug('subscriber for metric %s not found', metric_obj.local_id)
            else:
                logger.debug('Found subscribers for this metric, subscribing...: %s', this_metric_subscribers)
                # attach this metric id to discovered subscribers and select id <-> callbacks
                this_metric_subscribers['id'] = metric_obj.local_id
                found_callbacks = this_metric_subscribers[['id', 'callback']].set_index('id')
                # add this metric callbacks to DataManager's callbacks
                self.callbacks = self.callbacks.append(found_callbacks)
            return metric_obj
        else:
            raise NotImplementedError('Unknown metric type: %s' % _type)
1327915
def subscribe(self, session_id, dependencies=None):
        
        try:
            change_detection = self._meta.change_detection
            if change_detection not in [
                Options.CHANGE_DETECTION_PUSH,
                Options.CHANGE_DETECTION_POLL,
            ]:
                raise NotImplementedError(
                    "Change detection mechanism '{}' not implemented.".format(
                        change_detection
                    )
                )

            viewset_results = self._viewset_results()

            poll_interval = (
                self._meta.poll_interval
                if change_detection == Options.CHANGE_DETECTION_POLL
                else None
            )

            # Subscribe to observer in a single query. First, create an
            # observer, then create a subscriber, and finally subscribe to
            # the observer. If already subscribed, ignore the conflict.
            for retry in range(MAX_INTEGRITY_ERROR_RETRIES):
                is_subscribed = False
                cursor = connection.cursor()
                try:
                    cursor.execute(
                        """
                        WITH inserted_observer AS (
                            INSERT into {observer_table} ("id", "request", "poll_interval")
                            VALUES (%(observer_id)s, %(request)s, %(poll_interval)s)
                            ON CONFLICT DO NOTHING
                        ), inserted_subscriber AS (
                            INSERT into {subscriber_table} ("session_id", "created")
                            VALUES (%(subscriber_id)s, NOW())
                            ON CONFLICT DO NOTHING
                        )
                        INSERT INTO {observer_subscribers_table} ("observer_id", "subscriber_id")
                        VALUES (%(observer_id)s, %(subscriber_id)s)
                        """.format(
                            observer_table=models.Observer._meta.db_table,
                            subscriber_table=models.Subscriber._meta.db_table,
                            observer_subscribers_table=models.Observer.subscribers.through._meta.db_table,
                        ),
                        params={
                            'observer_id': self.id,
                            'request': pickle.dumps(self._request),
                            'poll_interval': poll_interval,
                            'subscriber_id': session_id,
                        },
                    )
                    is_subscribed = True
                except IntegrityError as err:
                    msg = str(err)
                    if (
                        'Key (observer_id, subscriber_id)' in msg
                        and 'already exists' in msg
                    ):
                        # Subscriber already subscribed, we're good.
                        is_subscribed = True
                    elif (
                        'Key (observer_id)' in msg or 'Key (subscriber_id)' in msg
                    ) and 'not present in table' in msg:
                        # Could not subscribe because observer, subscriber or
                        # both are missing, retry.
                        if retry == MAX_INTEGRITY_ERROR_RETRIES - 1:
                            raise
                    else:
                        raise
                finally:
                    cursor.close()

                if is_subscribed:
                    break

            # Determine who should notify us based on the configured change
            # detection mechanism.
            if change_detection == Options.CHANGE_DETECTION_PUSH:
                if dependencies:
                    tables = [model._meta.db_table for model in dependencies]
                else:
                    tables = [self._viewset.get_queryset().model._meta.db_table]

                # Register table dependencies for push observables.
                for table in tables:
                    try:
                        models.Dependency.objects.get_or_create(
                            observer_id=self.id, table=table
                        )
                    except models.Observer.DoesNotExist:
                        # The observer was removed before dependency tables
                        # were created.
                        return viewset_results

            elif self._meta.change_detection == Options.CHANGE_DETECTION_POLL:
                # Register poller.
                async_to_sync(get_channel_layer().send)(
                    CHANNEL_MAIN,
                    {
                        'type': TYPE_POLL,
                        'observer': self.id,
                        'interval': self._meta.poll_interval,
                    },
                )

            self._evaluate(viewset_results)

        except Exception:
            logger.exception(
                "Error while subscribing to observer ({})".format(
                    self._get_logging_id()
                ),
                extra=self._get_logging_extra(),
            )

        return viewset_results
1021351
def get_subscribers(self):
        
        return github.PaginatedList.PaginatedList(
            github.NamedUser.NamedUser,
            self._requester,
            self.url + "/subscribers",
            None
        )
366721
def subscribers(self):
        """"""
        return [s for p, s in sorted(self._subscribers, key=lambda x: x[0])]
897145
def run(self, service_id, **kwargs):
        
        log = self.get_logger(**kwargs)

        log.info("Loading Service for metric sync")
        try:
            service = Service.objects.get(id=service_id)
            log.info("Getting metrics for <%s>" % (service.name))
            metrics = self.get_metrics(service.url, service.token)
            result = metrics.json()
            if "metrics_available" in result:
                for key in result["metrics_available"]:
                    check = WidgetData.objects.filter(service=service, key=key)
                    if not check.exists():
                        WidgetData.objects.create(
                            service=service,
                            key=key,
                            title="TEMP - Pending update"
                        )
                        log.info("Add WidgetData for <%s>" % (key,))
            return "Completed metric sync for <%s>" % (service.name)
        except ObjectDoesNotExist:
            logger.error('Missing Service', exc_info=True)

        except SoftTimeLimitExceeded:
            logger.error(
                'Soft time limit exceed processing pull of service metrics \
                 via Celery.',
                exc_info=True)
49394
async def get_metrics(self):
        
        metrics = await self.model.get_metrics(self.tag)
        return metrics[self.name]
82960
def get_subscribers(self):
        
        data = self.signal.instance_subscribers
        if self.instance not in data:
            data[self.instance] = MethodAwareWeakList()
        return data[self.instance]
645858
def create(self):
        
        self.logger.debug('Create subscription on server...')

        if not self.connection.connected:
            self.state = 'connection_pending'
            return

        data = {
            'command': 'subscribe',
            'identifier': self._identifier_string()
        }

        self.connection.send(data)
        self.state = 'pending'
1236284
def create(self, callback_url):
        
        resource = self.resource.create({'subscribed_to': 'address',
                                         'callback_url': callback_url})
        subscription = self.wrap(resource)
        self.add(subscription)
        return subscription
632585
def process(self, metric):
        
        # Send the data as ......
        if not pika:
            return

        routingKeyDic = {
            'metric': lambda: metric.path,
            'custom': lambda: self.custom_routing_key,

            # These option and the below are really not needed because
            #  with Rabbitmq you can use regular expressions to indicate
            #  what routing_keys to subscribe to. But I figure this is
            #  a good example of how to allow more routing keys

            'host': lambda: metric.host,
            'metric.path': metric.getMetricPath,
            'path.prefix': metric.getPathPrefix,
            'collector.path': metric.getCollectorPath,
        }

        try:
            self.channel.basic_publish(
                exchange=self.topic_exchange,
                routing_key=routingKeyDic[self.routing_key](),
                body="%s" % metric)

        except Exception:  # Rough connection re-try logic.
            self.log.info(
                "Failed publishing to rabbitMQ. Attempting reconnect")
            self._bind()
1289435
def metrics(self, ticket):
        
        return self._query_zendesk(self.endpoint.metrics, 'ticket_metric', id=ticket)
990538
def subscribers(options = {})
      load
      return [] if info['push'].nil? || info['push']['subscribers'].nil?

      info['push']['subscribers'].map { |s| Subscriber.new(s, self, options) }
    end
1027242
def _publish_metrics(self, name, prev_keys, key, data):
        """"""
        value = data[key]
        keys = prev_keys + [key]
        if isinstance(value, dict):
            for new_key in value:
                self._publish_metrics(name, keys, new_key, value)
        elif isinstance(value, (float, int, long)):
            joined_keys = '.'.join(keys)
            if name:
                publish_key = '{}.{}'.format(name, joined_keys)
            else:
                publish_key = joined_keys
            if isinstance(value, bool):
                value = int(value)

            self.publish(publish_key, value)
1186454
def with_meter(name, tick_interval=meter.DEFAULT_TICK_INTERVAL):
    

    try:
        mmetric = new_meter(name, tick_interval)
    except DuplicateMetricError as e:
        mmetric = metric(name)

        if not isinstance(mmetric, meter.Meter):
            raise DuplicateMetricError("Metric {!r} already exists of type {}".format(name, type(mmetric).__name__))

        if tick_interval != mmetric.tick_interval:
            raise DuplicateMetricError("Metric {!r} already exists: {}".format(name, mmetric))

    def wrapper(f):

        @functools.wraps(f)
        def fun(*args, **kwargs):
            res = f(*args, **kwargs)

            mmetric.notify(1)
            return res

        return fun

    return wrapper
646890
def _resolve_metric(self, original_metric_name, metrics_to_collect, prefix=""):
        

        submit_method = (
            metrics_to_collect[original_metric_name][0]
            if isinstance(metrics_to_collect[original_metric_name], tuple)
            else metrics_to_collect[original_metric_name]
        )
        metric_name = (
            metrics_to_collect[original_metric_name][1]
            if isinstance(metrics_to_collect[original_metric_name], tuple)
            else original_metric_name
        )

        return submit_method, self._normalize(metric_name, submit_method, prefix)
571616
def meter(module, name, count=1):
    
    meter = get_metric('meters', module, name, Meter())
    meter.mark(count)
957433
def counter(self, key, **dims):
        """"""
        return super(MetricsRegistry, self).counter(
            self.metadata.register(key, **dims))