src/naarad/metrics/cluster_metric.py
def collect(self):
    """
    Take a list of metrics, filter all metrics based on hostname, and metric_type
    For each metric, merge the corresponding csv files into one,update corresponding properties such as csv_column_map.
    Users can specify functions: raw, count (qps), sum (aggregated value), avg (averaged value)
    The timestamp granularity of aggregated submetrics is in seconds (sub-second is not supported)
    """

    for aggr_metric in self.aggr_metrics:   # e.g., SAR-device.sda.await:count,sum,avg
      functions_aggr = []
      fields = aggr_metric.split(":")
      cur_metric_type = fields[0].split(".")[0]  # e.g. SAR-device

      if len(fields) > 1:  # The user has to specify the aggregate functions (i.e., :raw,count,sum,avg)
        func_user = ''.join(fields[1].split())
        functions_aggr.extend(func_user.split(","))
      else:  # no user input of aggregate functions
        return True

      cur_column = '.'.join(fields[0].split('.')[1:])    # e.g. sda.await or all.percent-sys

      # Store data points of various aggregation functions
      aggr_data = {}
      aggr_data['raw'] = []   # Store all the raw values
      aggr_data['sum'] = defaultdict(float)   # Store the sum values for each timestamp
      aggr_data['count'] = defaultdict(int)  # Store the count of each timestamp (i.e. qps)

      for metric in self.metrics:   # Loop the list to find from all metrics to merge
        if metric.hostname in self.aggr_hosts and \
           cur_column in metric.csv_column_map.values():
          file_csv = metric.get_csv(cur_column)
          timestamp_format = None
          with open(file_csv) as fh:
            for line in fh:
              aggr_data['raw'].append(line.rstrip())
              words = line.split(",")
              ts = words[0].split('.')[0]   # In case of sub-seconds; we only want the value of seconds;
              if not timestamp_format or timestamp_format == 'unknown':
                timestamp_format = naarad.utils.detect_timestamp_format(ts)
              if timestamp_format == 'unknown':
                continue
              ts = naarad.utils.get_standardized_timestamp(ts, timestamp_format)
              aggr_data['sum'][ts] += float(words[1])
              aggr_data['count'][ts] += 1
      # "raw" csv file
      if 'raw' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'raw')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          fh.write("\n".join(sorted(aggr_data['raw'])))

      # "sum"  csv file
      if 'sum' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'sum')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          for (k, v) in sorted(aggr_data['sum'].items()):
            fh.write(k + "," + str(v) + '\n')

      # "avg" csv file
      if 'avg' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'avg')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          for (k, v) in sorted(aggr_data['sum'].items()):
            fh.write(k + "," + str(v / aggr_data['count'][k]) + '\n')

      # "count" csv file (qps)
      if 'count' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'count')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          for (k, v) in sorted(aggr_data['count'].items()):
            fh.write(k + "," + str(v) + '\n')

      gc.collect()
    return True
def collect(self):
    """
    Take a list of metrics, filter all metrics based on hostname, and metric_type
    For each metric, merge the corresponding csv files into one,update corresponding properties such as csv_column_map.
    Users can specify functions: raw, count (qps), sum (aggregated value), avg (averaged value)
    The timestamp granularity of aggregated submetrics is in seconds (sub-second is not supported)
    """

    for aggr_metric in self.aggr_metrics:   # e.g., SAR-device.sda.await:count,sum,avg
      functions_aggr = []
      fields = aggr_metric.split(":")
      cur_metric_type = fields[0].split(".")[0]  # e.g. SAR-device

      if len(fields) > 1:  # The user has to specify the aggregate functions (i.e., :raw,count,sum,avg)
        func_user = ''.join(fields[1].split())
        functions_aggr.extend(func_user.split(","))
      else:  # no user input of aggregate functions
        return True

      cur_column = '.'.join(fields[0].split('.')[1:])    # e.g. sda.await or all.percent-sys

      # Store data points of various aggregation functions
      aggr_data = {}
      aggr_data['raw'] = []   # Store all the raw values
      aggr_data['sum'] = defaultdict(float)   # Store the sum values for each timestamp
      aggr_data['count'] = defaultdict(int)  # Store the count of each timestamp (i.e. qps)

      for metric in self.metrics:   # Loop the list to find from all metrics to merge
        if metric.hostname in self.aggr_hosts and \
           cur_column in metric.csv_column_map.values():
          file_csv = metric.get_csv(cur_column)
          timestamp_format = None
          with open(file_csv) as fh:
            for line in fh:
              aggr_data['raw'].append(line.rstrip())
              words = line.split(",")
              ts = words[0].split('.')[0]   # In case of sub-seconds; we only want the value of seconds;
              if not timestamp_format or timestamp_format == 'unknown':
                timestamp_format = naarad.utils.detect_timestamp_format(ts)
              if timestamp_format == 'unknown':
                continue
              ts = naarad.utils.get_standardized_timestamp(ts, timestamp_format)
              aggr_data['sum'][ts] += float(words[1])
              aggr_data['count'][ts] += 1
      # "raw" csv file
      if 'raw' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'raw')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          fh.write("\n".join(sorted(aggr_data['raw'])))

      # "sum"  csv file
      if 'sum' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'sum')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          for (k, v) in sorted(aggr_data['sum'].items()):
            fh.write(k + "," + str(v) + '\n')

      # "avg" csv file
      if 'avg' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'avg')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          for (k, v) in sorted(aggr_data['sum'].items()):
            fh.write(k + "," + str(v / aggr_data['count'][k]) + '\n')

      # "count" csv file (qps)
      if 'count' in functions_aggr:
        out_csv = self.get_csv(cur_column, 'count')
        self.csv_files.append(out_csv)
        with open(out_csv, 'w') as fh:
          for (k, v) in sorted(aggr_data['count'].items()):
            fh.write(k + "," + str(v) + '\n')

      gc.collect()
    return True
def collect ( self ) : for aggr_metric in self . aggr_metrics : # e.g., SAR-device.sda.await:count,sum,avg functions_aggr = [ ] fields = aggr_metric . split ( ":" ) cur_metric_type = fields [ 0 ] . split ( "." ) [ 0 ] # e.g. SAR-device if len ( fields ) > 1 : # The user has to specify the aggregate functions (i.e., :raw,count,sum,avg) func_user = '' . join ( fields [ 1 ] . split ( ) ) functions_aggr . extend ( func_user . split ( "," ) ) else : # no user input of aggregate functions return True cur_column = '.' . join ( fields [ 0 ] . split ( '.' ) [ 1 : ] ) # e.g. sda.await or all.percent-sys # Store data points of various aggregation functions aggr_data = { } aggr_data [ 'raw' ] = [ ] # Store all the raw values aggr_data [ 'sum' ] = defaultdict ( float ) # Store the sum values for each timestamp aggr_data [ 'count' ] = defaultdict ( int ) # Store the count of each timestamp (i.e. qps) for metric in self . metrics : # Loop the list to find from all metrics to merge if metric . hostname in self . aggr_hosts and cur_column in metric . csv_column_map . values ( ) : file_csv = metric . get_csv ( cur_column ) timestamp_format = None with open ( file_csv ) as fh : for line in fh : aggr_data [ 'raw' ] . append ( line . rstrip ( ) ) words = line . split ( "," ) ts = words [ 0 ] . split ( '.' ) [ 0 ] # In case of sub-seconds; we only want the value of seconds; if not timestamp_format or timestamp_format == 'unknown' : timestamp_format = naarad . utils . detect_timestamp_format ( ts ) if timestamp_format == 'unknown' : continue ts = naarad . utils . get_standardized_timestamp ( ts , timestamp_format ) aggr_data [ 'sum' ] [ ts ] += float ( words [ 1 ] ) aggr_data [ 'count' ] [ ts ] += 1 # "raw" csv file if 'raw' in functions_aggr : out_csv = self . get_csv ( cur_column , 'raw' ) self . csv_files . append ( out_csv ) with open ( out_csv , 'w' ) as fh : fh . write ( "\n" . join ( sorted ( aggr_data [ 'raw' ] ) ) ) # "sum"  csv file if 'sum' in functions_aggr : out_csv = self . get_csv ( cur_column , 'sum' ) self . csv_files . append ( out_csv ) with open ( out_csv , 'w' ) as fh : for ( k , v ) in sorted ( aggr_data [ 'sum' ] . items ( ) ) : fh . write ( k + "," + str ( v ) + '\n' ) # "avg" csv file if 'avg' in functions_aggr : out_csv = self . get_csv ( cur_column , 'avg' ) self . csv_files . append ( out_csv ) with open ( out_csv , 'w' ) as fh : for ( k , v ) in sorted ( aggr_data [ 'sum' ] . items ( ) ) : fh . write ( k + "," + str ( v / aggr_data [ 'count' ] [ k ] ) + '\n' ) # "count" csv file (qps) if 'count' in functions_aggr : out_csv = self . get_csv ( cur_column , 'count' ) self . csv_files . append ( out_csv ) with open ( out_csv , 'w' ) as fh : for ( k , v ) in sorted ( aggr_data [ 'count' ] . items ( ) ) : fh . write ( k + "," + str ( v ) + '\n' ) gc . collect ( ) return True
Take a list of metrics, filter all metrics based on hostname, and metric_type
    For each metric, merge the corresponding csv files into one,update corresponding properties such as csv_column_map.
    Users can specify functions: raw, count (qps), sum (aggregated value), avg (averaged value)
    The timestamp granularity of aggregated submetrics is in seconds (sub-second is not supported)
Take a list of metrics filter all metrics based on hostname and metric_type For each metric merge the corresponding csv files into one update corresponding properties such as csv_column_map . Users can specify functions : raw count ( qps ) sum ( aggregated value ) avg ( averaged value ) The timestamp granularity of aggregated submetrics is in seconds ( sub - second is not supported )