(self, tags)
Given a list of custom_queries, execute each query and parse the result for metrics
Given a list of custom_queries, execute each query and parse the result for metrics
Given a list of custom_queries execute each query and parse the result for metrics
def _collect_custom_queries(self, tags):
        """
        Given a list of custom_queries, execute each query and parse the result for metrics
        """
        for custom_query in self._config.custom_queries:
            metric_prefix = custom_query.get('metric_prefix')
            if not metric_prefix:
                self.log.error("custom query field `metric_prefix` is required")
                continue
            metric_prefix = metric_prefix.rstrip('.')

            query = custom_query.get('query')
            if not query:
                self.log.error("custom query field `query` is required for metric_prefix `%s`", metric_prefix)
                continue

            columns = custom_query.get('columns')
            if not columns:
                self.log.error("custom query field `columns` is required for metric_prefix `%s`", metric_prefix)
                continue

            cursor = self.db.cursor()
            with closing(cursor) as cursor:
                try:
                    self.log.debug("Running query: %s", query)
                    cursor.execute(query)
                except (psycopg2.ProgrammingError, psycopg2.errors.QueryCanceled) as e:
                    self.log.error("Error executing query for metric_prefix %s: %s", metric_prefix, str(e))
                    self.db.rollback()
                    continue

                for row in cursor:
                    if not row:
                        self.log.debug("query result for metric_prefix %s: returned an empty result", metric_prefix)
                        continue

                    if len(columns) != len(row):
                        self.log.error(
                            "query result for metric_prefix %s: expected %s columns, got %s",
                            metric_prefix,
                            len(columns),
                            len(row),
                        )
                        continue

                    metric_info = []
                    query_tags = list(custom_query.get('tags', []))
                    query_tags.extend(tags)

                    for column, value in zip(columns, row):
                        # Columns can be ignored via configuration.
                        if not column:
                            continue

                        name = column.get('name')
                        if not name:
                            self.log.error("column field `name` is required for metric_prefix `%s`", metric_prefix)
                            break

                        column_type = column.get('type')
                        if not column_type:
                            self.log.error(
                                "column field `type` is required for column `%s` of metric_prefix `%s`",
                                name,
                                metric_prefix,
                            )
                            break

                        if column_type == 'tag':
                            query_tags.append('{}:{}'.format(name, value))
                        else:
                            if not hasattr(self, column_type):
                                self.log.error(
                                    "invalid submission method `%s` for column `%s` of metric_prefix `%s`",
                                    column_type,
                                    name,
                                    metric_prefix,
                                )
                                break
                            try:
                                metric_info.append(('{}.{}'.format(metric_prefix, name), float(value), column_type))
                            except (ValueError, TypeError):
                                self.log.error(
                                    "non-numeric value `%s` for metric column `%s` of metric_prefix `%s`",
                                    value,
                                    name,
                                    metric_prefix,
                                )
                                break

                    # Only submit metrics if there were absolutely no errors - all or nothing.
                    else:
                        for info in metric_info:
                            metric, value, method = info
                            getattr(self, method)(metric, value, tags=set(query_tags), hostname=self.resolved_hostname)
def _collect_custom_queries ( self , tags ) : for custom_query in self . _config . custom_queries : metric_prefix = custom_query . get ( 'metric_prefix' ) if not metric_prefix : self . log . error ( "custom query field `metric_prefix` is required" ) continue metric_prefix = metric_prefix . rstrip ( '.' ) query = custom_query . get ( 'query' ) if not query : self . log . error ( "custom query field `query` is required for metric_prefix `%s`" , metric_prefix ) continue columns = custom_query . get ( 'columns' ) if not columns : self . log . error ( "custom query field `columns` is required for metric_prefix `%s`" , metric_prefix ) continue cursor = self . db . cursor ( ) with closing ( cursor ) as cursor : try : self . log . debug ( "Running query: %s" , query ) cursor . execute ( query ) except ( psycopg2 . ProgrammingError , psycopg2 . errors . QueryCanceled ) as e : self . log . error ( "Error executing query for metric_prefix %s: %s" , metric_prefix , str ( e ) ) self . db . rollback ( ) continue for row in cursor : if not row : self . log . debug ( "query result for metric_prefix %s: returned an empty result" , metric_prefix ) continue if len ( columns ) != len ( row ) : self . log . error ( "query result for metric_prefix %s: expected %s columns, got %s" , metric_prefix , len ( columns ) , len ( row ) , ) continue metric_info = [ ] query_tags = list ( custom_query . get ( 'tags' , [ ] ) ) query_tags . extend ( tags ) for column , value in zip ( columns , row ) : # Columns can be ignored via configuration. if not column : continue name = column . get ( 'name' ) if not name : self . log . error ( "column field `name` is required for metric_prefix `%s`" , metric_prefix ) break column_type = column . get ( 'type' ) if not column_type : self . log . error ( "column field `type` is required for column `%s` of metric_prefix `%s`" , name , metric_prefix , ) break if column_type == 'tag' : query_tags . append ( '{}:{}' . format ( name , value ) ) else : if not hasattr ( self , column_type ) : self . log . error ( "invalid submission method `%s` for column `%s` of metric_prefix `%s`" , column_type , name , metric_prefix , ) break try : metric_info . append ( ( '{}.{}' . format ( metric_prefix , name ) , float ( value ) , column_type ) ) except ( ValueError , TypeError ) : self . log . error ( "non-numeric value `%s` for metric column `%s` of metric_prefix `%s`" , value , name , metric_prefix , ) break # Only submit metrics if there were absolutely no errors - all or nothing. else : for info in metric_info : metric , value , method = info getattr ( self , method ) ( metric , value , tags = set ( query_tags ) , hostname = self . resolved_hostname )