yudaike/openSUSE-release-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
openSUSE-release-tools/metrics.py

591 lines
23 KiB
Python
Raw Normal View History

Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
#!/usr/bin/python
import argparse
from collections import namedtuple
from datetime import datetime
from dateutil.parser import parse as date_parse
from influxdb import InfluxDBClient
from lxml import etree as ET
import os
import subprocess
import sys
import yaml
metrics: provide release metrics ingestion and dashboard.
2018-03-08 21:54:29 -06:00
import metrics_release
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
import osc.conf
import osc.core
Catch 404 in metrics SP1 has no dashboard
2018-06-18 16:29:27 +02:00
from osc.core import HTTPError
metrics: provide dashboard revision walking function and cache patterns.
2018-04-20 16:30:27 -05:00
from osc.core import get_commitlog
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
import osclib.conf
from osclib.cache import Cache
from osclib.conf import Config
Replace relevant 'dashboard' package references with pseudometa lookup. Skipping some of the broader once which have no tangible impact, but are expensive to change.
2018-08-17 22:23:09 -05:00
from osclib.core import project_pseudometa_package
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
from osclib.stagingapi import StagingAPI
metrics: prefix release schedule file with source dir path. Otherwise, when deployed via package and run via /usr/bin the files are not found.
2017-11-08 16:42:42 -06:00
SOURCE_DIR = os.path.dirname(os.path.realpath(__file__))
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
Point = namedtuple('Point', ['measurement', 'tags', 'fields', 'time', 'delta'])
metrics: prefix release schedule file with source dir path. Otherwise, when deployed via package and run via /usr/bin the files are not found.
2017-11-08 16:42:42 -06:00
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
# Duplicate Leap config to handle 13.2 without issue.
Another set of low-noise flake8 cleanups
2017-10-20 08:54:37 +02:00
osclib.conf.DEFAULT[
metrics: correct reference to Leap config defaults pattern.
2018-09-05 14:24:51 -05:00
r'openSUSE:(?P<project>[\d.]+)$'] = osclib.conf.DEFAULT[
Remove nonfree handling Prepare NonFree as onlyadi project, so it can have its own staging workflow
2019-11-21 13:42:19 +01:00
r'openSUSE:(?P<project>Leap:(?P<version>[\d.]+))$']
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
# Provide osc.core.get_request_list() that swaps out search() implementation to
# capture the generated query, paginate over and yield each request to avoid
# loading all requests at the same time. Additionally, use lxml ET to avoid
# having to re-parse to perform complex xpaths.
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
def get_request_list(*args, **kwargs):
osc.core._search = osc.core.search
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
osc.core.search = search_capture
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
osc.core._ET = osc.core.ET
osc.core.ET = ET
metrics: remove workaround for OBS xpath bug during request search. Now that the bug related to openSUSE/open-build-service#5571 has been fixed the inverted behavior of exclude_target_projects no longer works. As such the argument should be removed and only way to get the desired behavior is to override include_request_from_project which includes request sourced from specific project. For the purposes of metrics only interested in requests targeting a specific project.
2018-09-08 22:15:43 -05:00
osc.conf.config['include_request_from_project'] = False
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
osc.core.get_request_list(*args, **kwargs)
osc.core.search = osc.core._search
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
query = search_capture.query
for request in search_paginated_generator(query[0], query[1], **query[2]):
# Python 3 yield from.
yield request
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
osc.core.ET = osc.core._ET
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
def search_capture(apiurl, queries=None, **kwargs):
search_capture.query = (apiurl, queries, kwargs)
return {'request': ET.fromstring('<collection matches="0"></collection>')}
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
# Provides a osc.core.search() implementation for use with get_request_list()
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
# that paginates in sets of 1000 and yields each request.
def search_paginated_generator(apiurl, queries=None, **kwargs):
metrics: check for action/target instead of submit/target since removed. In the wake of openSUSE/osc@f1c3156 removing the deprecated submit/target, the special condition to limit Factory to "staging era" requests should be updated to look for action/target.
2018-09-08 22:17:56 -05:00
if "action/target/@project='openSUSE:Factory'" in kwargs['request']:
metrics: reduce request period to after 450000. Works around ingest job crashing due to lack of memory. Hard to properly workaround due to lack of sort API option from OBS.
2019-05-09 15:36:01 -05:00
# Idealy this would be 250000, but poo#48437 and lack of OBS sort.
kwargs['request'] = osc.core.xpath_join(kwargs['request'], '@id>450000', op='and')
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
request_count = 0
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
queries['request']['limit'] = 1000
queries['request']['offset'] = 0
while True:
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
collection = osc.core.search(apiurl, queries, **kwargs)['request']
if not request_count:
print('processing {:,} requests'.format(int(collection.get('matches'))))
for request in collection.findall('request'):
yield request
request_count += 1
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
metrics: rework request pagination to provide as generator. This prevents the memory required to process to be enough to load all parsed request element trees at once. Instead only one page of requests is loaded at a time and the memory freed after processed. The end result is the memory consumption reduced by just over 20% (current Factory drops by around 2.5GB).
2017-11-09 01:23:03 -06:00
if request_count == int(collection.get('matches')):
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
# Stop paging once the expected number of items has been returned.
break
metrics: call ET.clear() to release unneeded memory used by search result. Roughly 1800MB per 10,000 requests saved.
2017-11-09 01:17:27 -06:00
# Release memory as otherwise ET seems to hold onto it.
collection.clear()
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
queries['request']['offset'] += queries['request']['limit']
points = []
def point(measurement, fields, datetime, tags={}, delta=False):
global points
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
points.append(Point(measurement, tags, fields, timestamp(datetime), delta))
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
def timestamp(datetime):
return int(datetime.strftime('%s'))
def ingest_requests(api, project):
requests = get_request_list(api.apiurl, project,
req_state=('accepted', 'revoked', 'superseded'),
withfullhistory=True)
for request in requests:
if request.find('action').get('type') not in ('submit', 'delete'):
# TODO Handle non-stageable requests via different flow.
continue
created_at = date_parse(request.find('history').get('when'))
final_at = date_parse(request.find('state').get('when'))
final_at_history = date_parse(request.find('history[last()]').get('when'))
if final_at_history > final_at:
# Workaround for invalid dates: openSUSE/open-build-service#3858.
final_at = final_at_history
# TODO Track requests in psuedo-ignore state.
point('total', {'backlog': 1, 'open': 1}, created_at, {'event': 'create'}, True)
point('total', {'backlog': -1, 'open': -1}, final_at, {'event': 'close'}, True)
request_tags = {}
request_fields = {
'total': (final_at - created_at).total_seconds(),
'staged_count': len(request.findall('review[@by_group="factory-staging"]/history')),
}
# TODO Total time spent in backlog (ie factory-staging, but excluding when staged).
staged_first_review = request.xpath('review[contains(@by_project, "{}:Staging:")]'.format(project))
if len(staged_first_review):
by_project = staged_first_review[0].get('by_project')
request_tags['type'] = 'adi' if api.is_adi_project(by_project) else 'letter'
# TODO Determine current whitelists state based on dashboard revisions.
if project.startswith('openSUSE:Factory'):
splitter_whitelist = 'B C D E F G H I J'.split()
if splitter_whitelist:
short = api.extract_staging_short(by_project)
request_tags['whitelisted'] = short in splitter_whitelist
else:
# All letter where whitelisted since no restriction.
request_tags['whitelisted'] = request_tags['type'] == 'letter'
ready_to_accept = request.xpath('review[contains(@by_project, "{}:Staging:adi:") and @state="accepted"]/history[comment[text() = "ready to accept"]]/@when'.format(project))
if len(ready_to_accept):
ready_to_accept = date_parse(ready_to_accept[0])
request_fields['ready'] = (final_at - ready_to_accept).total_seconds()
# TODO Points with indentical timestamps are merged so this can be placed in total
# measurement, but may make sense to keep this separate and make the others follow.
point('ready', {'count': 1}, ready_to_accept, delta=True)
point('ready', {'count': -1}, final_at, delta=True)
staged_first = request.xpath('review[@by_group="factory-staging"]/history/@when')
if len(staged_first):
staged_first = date_parse(staged_first[0])
request_fields['staged_first'] = (staged_first - created_at).total_seconds()
# TODO Decide if better to break out all measurements by time most relevant to event,
# time request was created, or time request was finalized. It may also make sense to
# keep separate measurement by different times like this one.
point('request_staged_first', {'value': request_fields['staged_first']}, staged_first, request_tags)
point('request', request_fields, final_at, request_tags)
# Staging related reviews.
for number, review in enumerate(
request.xpath('review[contains(@by_project, "{}:Staging:")]'.format(project)), start=1):
staged_at = date_parse(review.get('when'))
project_type = 'adi' if api.is_adi_project(review.get('by_project')) else 'letter'
short = api.extract_staging_short(review.get('by_project'))
point('staging', {'count': 1}, staged_at,
{'id': short, 'type': project_type, 'event': 'select'}, True)
point('total', {'backlog': -1, 'staged': 1}, staged_at, {'event': 'select'}, True)
who = who_workaround(request, review)
review_tags = {'event': 'select', 'user': who, 'number': number}
review_tags.update(request_tags)
point('user', {'count': 1}, staged_at, review_tags)
history = review.find('history')
if history is not None:
unselected_at = date_parse(history.get('when'))
else:
unselected_at = final_at
# If a request is declined and re-opened it must be repaired before being re-staged. At
# which point the only possible open review should be the final one.
point('staging', {'count': -1}, unselected_at,
{'id': short, 'type': project_type, 'event': 'unselect'}, True)
point('total', {'backlog': 1, 'staged': -1}, unselected_at, {'event': 'unselect'}, True)
# No-staging related reviews.
for review in request.xpath('review[not(contains(@by_project, "{}:Staging:"))]'.format(project)):
tags = {
# who_added is non-trivial due to openSUSE/open-build-service#3898.
'state': review.get('state'),
}
opened_at = date_parse(review.get('when'))
history = review.find('history')
if history is not None:
completed_at = date_parse(history.get('when'))
tags['who_completed'] = history.get('who')
else:
completed_at = final_at
# Does not seem to make sense to mirror user responsible for making final state
# change as the user who completed the review.
tags['key'] = []
tags['type'] = []
for name, value in sorted(review.items(), reverse=True):
if name.startswith('by_'):
tags[name] = value
tags['key'].append(value)
tags['type'].append(name[3:])
tags['type'] = '_'.join(tags['type'])
point('review', {'open_for': (completed_at - opened_at).total_seconds()}, completed_at, tags)
point('review_count', {'count': 1}, opened_at, tags, True)
point('review_count', {'count': -1}, completed_at, tags, True)
found = []
for set_priority in request.xpath('history[description[contains(text(), "Request got a new priority:")]]'):
parts = set_priority.find('description').text.rsplit(' ', 3)
priority_previous = parts[1]
priority = parts[3]
if priority == priority_previous:
continue
changed_at = date_parse(set_priority.get('when'))
if priority_previous != 'moderate':
point('priority', {'count': -1}, changed_at, {'level': priority_previous}, True)
if priority != 'moderate':
point('priority', {'count': 1}, changed_at, {'level': priority}, True)
found.append(priority)
# Ensure a final removal entry is created when request is finalized.
priority = request.find('priority')
if priority is not None and priority.text != 'moderate':
if priority.text in found:
point('priority', {'count': -1}, final_at, {'level': priority.text}, True)
else:
print('unable to find priority history entry for {} to {}'.format(request.get('id'), priority.text))
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
print('finalizing {:,} points'.format(len(points)))
return walk_points(points, project)
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
def who_workaround(request, review, relax=False):
# Super ugly workaround for incorrect and missing data:
# - openSUSE/open-build-service#3857
# - openSUSE/open-build-service#3898
global who_workaround_swap, who_workaround_miss
who = review.get('who') # All that should be required (used as fallback).
when = review.get('when')
if relax:
# Super hack, chop off seconds to relax in hopes of finding potential.
when = when[:-2]
who_real = request.xpath(
'history[contains(@when, "{}") and comment[contains(text(), "{}")]]/@who'.format(
when, review.get('by_project')))
if len(who_real):
who = who_real[0]
who_workaround_swap += 1
elif not relax:
return who_workaround(request, review, True)
else:
who_workaround_miss += 1
return who
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
# Walk data points in order by time, adding up deltas and merging points at
# the same time. Data is converted to dict() and written to influx batches to
# avoid extra memory usage required for all data in dict() and avoid influxdb
# allocating memory for entire incoming data set at once.
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
def walk_points(points, target):
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
global client
metrics: do not drop entire database, but instead measurements. This ensures that release data, which is ingest separately, is not wiped out. Additionally, it is less destructive during update.
2018-03-09 16:53:26 -06:00
measurements = set()
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
counters = {}
final = []
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
time_last = None
wrote = 0
for point in sorted(points, key=lambda l: l.time):
metrics: do not drop entire database, but instead measurements. This ensures that release data, which is ingest separately, is not wiped out. Additionally, it is less destructive during update.
2018-03-09 16:53:26 -06:00
if point.measurement not in measurements:
# Wait until just before writing to drop measurement.
client.drop_measurement(point.measurement)
measurements.add(point.measurement)
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
if point.time != time_last and len(final) >= 1000:
# Write final point in batches of ~1000, but guard against writing
# when in the middle of points at the same time as they may end up
# being merged. As such the previous time should not match current.
client.write_points(final, 's')
wrote += len(final)
final = []
time_last = point.time
if not point.delta:
final.append(dict(point._asdict()))
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
continue
# A more generic method like 'key' which ended up being needed is likely better.
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
measurement = counters_tag_key = point.measurement
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
if measurement == 'staging':
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
counters_tag_key += point.tags['id']
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
elif measurement == 'review_count':
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
counters_tag_key += '_'.join(point.tags['key'])
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
elif measurement == 'priority':
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
counters_tag_key += point.tags['level']
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
counters_tag = counters.setdefault(counters_tag_key, {'last': None, 'values': {}})
values = counters_tag['values']
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
for key, value in point.fields.items():
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
values[key] = values.setdefault(key, 0) + value
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
if counters_tag['last'] and point.time == counters_tag['last']['time']:
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
point = counters_tag['last']
else:
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
point = dict(point._asdict())
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
counters_tag['last'] = point
final.append(point)
point['fields'].update(counters_tag['values'])
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
# Write any remaining final points.
client.write_points(final, 's')
return wrote + len(final)
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
def ingest_release_schedule(project):
points = []
release_schedule = {}
metrics: prefix release schedule file with source dir path. Otherwise, when deployed via package and run via /usr/bin the files are not found.
2017-11-08 16:42:42 -06:00
release_schedule_file = os.path.join(SOURCE_DIR, 'metrics/annotation/{}.yaml'.format(project))
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
if project.endswith('Factory'):
metrics: ingest_release_schedule(): disable for Tumbleweed until #1250.
2018-02-19 14:40:17 -06:00
# TODO Pending resolution to #1250 regarding deployment.
return 0
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
# Extract Factory "release schedule" from Tumbleweed snapshot list.
command = 'rsync rsync.opensuse.org::opensuse-full/opensuse/tumbleweed/iso/Changes.* | ' \
flake8: utilize raw strings to avoid invalid escape character warnings.
2018-10-24 16:29:15 -05:00
'grep -oP "' + r'Changes\.\K\d{5,}' + '"'
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
snapshots = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE).communicate()[0]
for date in snapshots.split():
release_schedule[datetime.strptime(date, '%Y%m%d')] = 'Snapshot {}'.format(date)
elif os.path.isfile(release_schedule_file):
# Load release schedule for non-rolling releases from yaml file.
with open(release_schedule_file, 'r') as stream:
release_schedule = yaml.safe_load(stream)
for date, description in release_schedule.items():
points.append({
'measurement': 'release_schedule',
'fields': {'description': description},
'time': timestamp(date),
})
metrics: drop release schedule before writing. After change from dropping entire database to individual measurements this was overlooked and instead new points are simply written each time which results in duplicates.
2018-05-18 14:20:26 -05:00
client.drop_measurement('release_schedule')
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
client.write_points(points, 's')
return len(points)
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
metrics: provide dashboard revision walking function and cache patterns.
2018-04-20 16:30:27 -05:00
def revision_index(api):
if not hasattr(revision_index, 'index'):
revision_index.index = {}
Catch 404 in metrics SP1 has no dashboard
2018-06-18 16:29:27 +02:00
Replace relevant 'dashboard' package references with pseudometa lookup. Skipping some of the broader once which have no tangible impact, but are expensive to change.
2018-08-17 22:23:09 -05:00
project, package = project_pseudometa_package(api.apiurl, api.project)
Catch 404 in metrics SP1 has no dashboard
2018-06-18 16:29:27 +02:00
try:
Utilize ET.fromstringlist() intead of joining strings.
2018-08-16 00:26:53 -05:00
root = ET.fromstringlist(
Replace relevant 'dashboard' package references with pseudometa lookup. Skipping some of the broader once which have no tangible impact, but are expensive to change.
2018-08-17 22:23:09 -05:00
get_commitlog(api.apiurl, project, package, None, format='xml'))
Catch 404 in metrics SP1 has no dashboard
2018-06-18 16:29:27 +02:00
except HTTPError as e:
return revision_index.index
metrics: provide dashboard revision walking function and cache patterns.
2018-04-20 16:30:27 -05:00
for logentry in root.findall('logentry'):
date = date_parse(logentry.find('date').text)
revision_index.index[date] = logentry.get('revision')
return revision_index.index
def revision_at(api, datetime):
index = revision_index(api)
for made, revision in sorted(index.items(), reverse=True):
if made <= datetime:
return revision
return None
def dashboard_at(api, filename, datetime=None, revision=None):
if datetime:
revision = revision_at(api, datetime)
if not revision:
return revision
osclib/stagingapi: rename dashboard_content_* methods to pseudometa_file_*. No need to be confusing and use the deprecated dashboard name.
2018-08-17 23:19:24 -05:00
content = api.pseudometa_file_load(filename, revision)
metrics: provide dashboard revision walking function and cache patterns.
2018-04-20 16:30:27 -05:00
if filename in ('ignored_requests'):
if content:
return yaml.safe_load(content)
return {}
elif filename in ('config'):
if content:
# TODO re-use from osclib.conf.
from ConfigParser import ConfigParser
import io
cp = ConfigParser()
config = '[remote]\n' + content
cp.readfp(io.BytesIO(config))
return dict(cp.items('remote'))
return {}
return content
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
def dashboard_at_changed(api, filename, revision=None):
if not hasattr(dashboard_at_changed, 'previous'):
dashboard_at_changed.previous = {}
content = dashboard_at(api, filename, revision=revision)
if content is None and filename == 'repo_checker' and api.project == 'openSUSE:Factory':
# Special case to fallback to installcheck pre repo_checker file.
return dashboard_at_changed(api, 'installcheck', revision)
if content and content != dashboard_at_changed.previous.get(filename):
dashboard_at_changed.previous[filename] = content
return content
return None
def ingest_dashboard_config(content):
metrics: ingest_dashboard_config: rename seen to previous. More accurately describes the new purpose.
2018-04-24 18:20:44 -05:00
if not hasattr(ingest_dashboard_config, 'previous'):
metrics: ingest_dashboard_config: initialize seen with last record keys.
2018-04-24 18:19:30 -05:00
result = client.query('SELECT * FROM dashboard_config ORDER BY time DESC LIMIT 1')
if result:
# Extract last point and remove zero values since no need to fill.
point = next(result.get_points())
point = {k: v for (k, v) in point.iteritems() if k != 'time' and v != 0}
metrics: ingest_dashboard_config: rename seen to previous. More accurately describes the new purpose.
2018-04-24 18:20:44 -05:00
ingest_dashboard_config.previous = set(point.keys())
metrics: ingest_dashboard_config: initialize seen with last record keys.
2018-04-24 18:19:30 -05:00
else:
metrics: ingest_dashboard_config: rename seen to previous. More accurately describes the new purpose.
2018-04-24 18:20:44 -05:00
ingest_dashboard_config.previous = set()
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
fields = {}
for key, value in content.items():
if key.startswith('repo_checker-binary-whitelist'):
metrics: ingest_dashboard_config: rename seen to previous. More accurately describes the new purpose.
2018-04-24 18:20:44 -05:00
ingest_dashboard_config.previous.add(key)
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
fields[key] = len(value.split())
# Ensure any previously seen key are filled with zeros if no longer present
# to allow graphs to fill with previous.
metrics: ingest_dashboard_config: only fill missing values once.
2018-04-24 18:18:57 -05:00
fields_keys = set(fields.keys())
metrics: ingest_dashboard_config: rename seen to previous. More accurately describes the new purpose.
2018-04-24 18:20:44 -05:00
missing = ingest_dashboard_config.previous - fields_keys
metrics: ingest_dashboard_config: only fill missing values once.
2018-04-24 18:18:57 -05:00
if len(missing):
metrics: ingest_dashboard_config: rename seen to previous. More accurately describes the new purpose.
2018-04-24 18:20:44 -05:00
ingest_dashboard_config.previous = fields_keys
metrics: ingest_dashboard_config: only fill missing values once.
2018-04-24 18:18:57 -05:00
for key in missing:
fields[key] = 0
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
return fields
def ingest_dashboard_devel_projects(content):
return {
'count': len(content.strip().split()),
}
def ingest_dashboard_repo_checker(content):
return {
'install_count': content.count("can't install "),
'conflict_count': content.count('found conflict of '),
'line_count': content.count('\n'),
}
def ingest_dashboard_version_snapshot(content):
return {
'version': content.strip(),
}
def ingest_dashboard_revision_get():
metrics: dashboard: drop dashboard measurement in favor of dashboard_revision.
2018-04-24 19:27:18 -05:00
result = client.query('SELECT revision FROM dashboard_revision ORDER BY time DESC LIMIT 1')
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
if result:
return next(result.get_points())['revision']
return None
def ingest_dashboard(api):
index = revision_index(api)
revision_last = ingest_dashboard_revision_get()
past = True if revision_last is None else False
print('dashboard ingest: processing {:,} revisions starting after {}'.format(
len(index), 'the beginning' if past else revision_last))
filenames = ['config', 'repo_checker', 'version_snapshot']
if api.project == 'openSUSE:Factory':
filenames.append('devel_projects')
count = 0
points = []
for made, revision in sorted(index.items()):
if not past:
if revision == revision_last:
past = True
continue
time = timestamp(made)
for filename in filenames:
content = dashboard_at_changed(api, filename, revision)
if content:
map_func = globals()['ingest_dashboard_{}'.format(filename)]
fields = map_func(content)
if not len(fields):
continue
points.append({
'measurement': 'dashboard_{}'.format(filename),
'fields': fields,
'time': time,
})
metrics: dashboard: include revisions graph.
2018-04-24 19:24:52 -05:00
points.append({
'measurement': 'dashboard_revision',
'fields': {
'revision': revision,
},
'time': time,
})
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
if len(points) >= 1000:
client.write_points(points, 's')
count += len(points)
points = []
if len(points):
client.write_points(points, 's')
count += len(points)
metrics: ingest_dashboard(): replace max_revision duplicate variable nonsense Replaces commit 437f46a4cb156d44baf9f2c18a4d702240936836.
2018-06-20 15:21:52 -05:00
print('last revision processed: {}'.format(revision if len(index) else 'none'))
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
return count
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
def main(args):
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
global client
client = InfluxDBClient(args.host, args.port, args.user, args.password, args.project)
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
osc.conf.get_config(override_apiurl=args.apiurl)
Replace relevant 'dashboard' package references with pseudometa lookup. Skipping some of the broader once which have no tangible impact, but are expensive to change.
2018-08-17 22:23:09 -05:00
apiurl = osc.conf.config['apiurl']
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
osc.conf.config['debug'] = args.debug
metrics: do not drop entire database, but instead measurements. This ensures that release data, which is ingest separately, is not wiped out. Additionally, it is less destructive during update.
2018-03-09 16:53:26 -06:00
# Ensure database exists.
client.create_database(client._database)
metrics: provide release metrics ingestion and dashboard.
2018-03-08 21:54:29 -06:00
metrics_release.ingest(client)
if args.release_only:
return
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
# Use separate cache since it is persistent.
Replace relevant 'dashboard' package references with pseudometa lookup. Skipping some of the broader once which have no tangible impact, but are expensive to change.
2018-08-17 22:23:09 -05:00
_, package = project_pseudometa_package(apiurl, args.project)
metrics: provide --wipe-cache option.
2018-01-10 20:11:46 -06:00
if args.wipe_cache:
Cache.delete_all()
metrics: instead of wiping cache place ephemeral patterns behind flag. This allows for non-ephemeral queries to be cached indeinitely in the future (like source queries that include the revision). For development one may use --heavy-cache to be able to quickly iterate.
2018-04-20 16:25:47 -05:00
if args.heavy_cache:
flake8: utilize raw strings to avoid invalid escape character warnings.
2018-10-24 16:29:15 -05:00
Cache.PATTERNS[r'/search/request'] = sys.maxint
Cache.PATTERNS[r'/source/[^/]+/{}/_history'.format(package)] = sys.maxint
Cache.PATTERNS[r'/source/[^/]+/{}/[^/]+\?rev=.*'.format(package)] = sys.maxint
Utilize CacheManager for all existing caches.
2018-09-04 15:03:03 -05:00
Cache.init('metrics')
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
osclib/conf: decouple from StagingAPI and always fetch remote config. As the remote config is no longer optional for SLE and is utilized by openSUSE to the point were it is dangerous not to load the remote config it should be required. Currently only certain users call apply_remote() while this will make it built-in during construction and thus makes the usage consistent and no longer require StagingAPI.
2018-08-16 21:46:05 -05:00
Config(apiurl, args.project)
api = StagingAPI(apiurl, args.project)
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
metrics: ingest relevant dashboard files and extract metrics.
2018-04-20 16:32:02 -05:00
print('dashboard: wrote {:,} points'.format(ingest_dashboard(api)))
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
global who_workaround_swap, who_workaround_miss
who_workaround_swap = who_workaround_miss = 0
points_requests = ingest_requests(api, args.project)
points_schedule = ingest_release_schedule(args.project)
print('who_workaround_swap', who_workaround_swap)
print('who_workaround_miss', who_workaround_miss)
metrics: rework to store points as named tuple and write in batches. Savings of around 400MB per 10,000 requests. Using a named tuple was the original approach for this reason, but the influxdb interface requires dict()s and it seemed silly to spend time converting them. Additionally, influxdb client already does batching. Unfortunately, with the amount of data processed for Factory that will continue to grow this approach is necessary. The dict() final structures are buffered up to ~1000 before being written and released. Another benefit of the batching is that influxdb does not allocate memory for the entire incoming batch.
2017-11-09 01:29:57 -06:00
print('wrote {:,} points and {:,} annotation points to db'.format(
points_requests, points_schedule))
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
if __name__ == '__main__':
description = 'Ingest relevant OBS and annotation data to generate insightful metrics.'
parser = argparse.ArgumentParser(description=description)
parser.add_argument('-A', '--apiurl', help='OBS instance API URL')
parser.add_argument('-d', '--debug', action='store_true', help='print useful debugging info')
parser.add_argument('-p', '--project', default='openSUSE:Factory', help='OBS project')
parser.add_argument('--host', default='localhost', help='InfluxDB host')
parser.add_argument('--port', default=8086, help='InfluxDB post')
parser.add_argument('--user', default='root', help='InfluxDB user')
parser.add_argument('--password', default='root', help='InfluxDB password')
metrics: provide --wipe-cache option.
2018-01-10 20:11:46 -06:00
parser.add_argument('--wipe-cache', action='store_true', help='wipe GET request cache before executing')
metrics: instead of wiping cache place ephemeral patterns behind flag. This allows for non-ephemeral queries to be cached indeinitely in the future (like source queries that include the revision). For development one may use --heavy-cache to be able to quickly iterate.
2018-04-20 16:25:47 -05:00
parser.add_argument('--heavy-cache', action='store_true',
help='cache ephemeral queries indefinitely (useful for development)')
metrics: provide release metrics ingestion and dashboard.
2018-03-08 21:54:29 -06:00
parser.add_argument('--release-only', action='store_true', help='ingest release metrics only')
Initial metrics ingestion, processing, write to influxdb, and grafana dashboards. A lot of room for improvement and additional metrics that can be extracted. Including non-final state requests would allow for analyzing the current staging state instead of only historical state. Additionally, the current state can be used to present an activity log. Handling incremental updates is non-trivial given the deltas are evaluated and stored in sum state. A few possible approaches, but likely not worth the hassle given the relatively short processing time and infrequent desire to update data (daily at minimum).
2017-10-04 00:58:43 -05:00
args = parser.parse_args()
sys.exit(main(args))
Reference in New Issue Copy Permalink