def timeseries()

in blueprints/cloud-operations/network-quota-monitoring/src/plugins/series-subnets.py [0:0]

• 30 lines of code
• 9 McCabe index (conditional complexity)

def timeseries(resources):
  'Returns used/available/ratio timeseries for addresses by subnetwork.'
  LOGGER.info('timeseries')
  # return descriptors
  for dtype, name in DESCRIPTOR_ATTRS.items():
    yield MetricDescriptor(f'subnetwork/{dtype}', name,
                           ('project', 'network', 'subnetwork', 'region'),
                           dtype.endswith('ratio'))
  # aggregate per-resource counts in total per-subnet counts
  subnet_nets = {
      k: ipaddress.ip_network(v['cidr_range'])
      for k, v in resources['subnetworks'].items()
  }
  # TODO: add counter functions for PSA
  subnet_counts = {k: 0 for k in resources['subnetworks']}
  counters = itertools.chain(_subnet_addresses(resources),
                             _subnet_forwarding_rules(resources, subnet_nets),
                             _subnet_instances(resources))
  for subnet_self_link, count in counters:
    subnet_counts[subnet_self_link] += count
  # compute and return metrics
  for subnet_self_link, count in subnet_counts.items():
    max_ips = subnet_nets[subnet_self_link].num_addresses - 4
    subnet = resources['subnetworks'][subnet_self_link]
    labels = {
        'network': resources['networks'][subnet['network']]['name'],
        'project': subnet['project_id'],
        'region': subnet['region'],
        'subnetwork': subnet['name']
    }
    yield TimeSeries('subnetwork/addresses_available', max_ips, labels)
    yield TimeSeries('subnetwork/addresses_used', count, labels)
    yield TimeSeries('subnetwork/addresses_used_ratio',
                     0 if count == 0 else count / max_ips, labels)