opendev/base-jobs
Code Issues Proposed changes

Stop submitting logstash/subunit gearman jobs

Browse Source 
The openstack health server stopped working a few months ago and we
ended up shutting down the subunit workers and the health api server as
a result. This means we can stop submitting gearman jobs to process
subunit files.

Also about a year ago we indicated to OpenStack that we could keep the
logstash tooling running through the yoga cycle which is now over. We
haven't had any volunteers or help to continue running the ELK stack in
opendev so we're going to shut it down now that yoga is out the door.
Openstack did end up working with AWS to set up an opensearch
replacement which users can look to for log indexing of CI jobs in
OpenStack.

Change-Id: I5f0f3805e191f0cd6354285299ed33c42d3899fd
This commit is contained in:
Clark Boylan 2022-04-12 12:59:42 -07:00
parent 17a4e88204
commit 9078f71fbb
15 changed files with 0 additions and 4335 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
playbooks/base-test/post-logs.yaml
View File

@ -37,11 +37,3 @@
url: 'download-logs.sh'
metadata:
command: 'curl "{{ upload_results.url }}/download-logs.sh" | bash'
- hosts: localhost
# NOTE(pabelanger): We ignore_errors for the following tasks as not to fail
# successful jobs.
ignore_errors: yes
roles:
- submit-logstash-jobs
- submit-subunit-jobs

8
playbooks/base/post-logs.yaml
View File

@ -37,11 +37,3 @@
url: 'download-logs.sh'
metadata:
command: 'curl "{{ upload_results.url }}/download-logs.sh" | bash'
- hosts: localhost
# NOTE(pabelanger): We ignore_errors for the following tasks as not to fail
# successful jobs.
ignore_errors: yes
roles:
- submit-logstash-jobs
- submit-subunit-jobs

6
roles/submit-log-processor-jobs/README.rst
View File

@ -1,6 +0,0 @@
A module to submit a log processing job.
This role is a container for an Ansible module which processes a log
directory and submits jobs to a log processing gearman queue. The
role itself performs no actions, and is intended only to be used by
other roles as a dependency to supply the module.

212
roles/submit-log-processor-jobs/library/submit_log_processor_jobs.py
View File

@ -1,212 +0,0 @@
# Copyright 2013 Hewlett-Packard Development Company, L.P.
# Copyright (C) 2017 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
#
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import json
import re
import traceback
from ansible.module_utils.six.moves import urllib
from ansible.module_utils.basic import AnsibleModule
import ansible.module_utils.gear as gear
class FileMatcher(object):
def __init__(self, name, tags):
self._name = name
self.name = re.compile(name)
self.tags = tags
def matches(self, s):
if self.name.search(s):
return True
class File(object):
def __init__(self, name, tags):
self._name = name
self._tags = tags
@property
def name(self):
return self._name
@name.setter
def name(self, value):
raise Exception("Cannot update File() objects they must be hashable")
@property
def tags(self):
return self._tags
@tags.setter
def tags(self, value):
raise Exception("Cannot update File() objects they must be hashable")
def toDict(self):
return dict(name=self.name,
tags=self.tags)
# We need these objects to be hashable so that we can use sets
# below.
def __eq__(self, other):
return self.name == other.name
def __ne__(self, other):
return not self.__eq__(other)
def __hash__(self):
return hash(self.name)
class LogMatcher(object):
def __init__(self, server, port, config, success, log_url, host_vars):
self.client = gear.Client()
self.client.addServer(server, port)
self.hosts = host_vars
self.zuul = list(host_vars.values())[0]['zuul']
self.success = success
self.log_url = log_url
self.matchers = []
for f in config['files']:
self.matchers.append(FileMatcher(f['name'], f.get('tags', [])))
def findFiles(self, path):
results = set()
for (dirpath, dirnames, filenames) in os.walk(path):
for filename in filenames:
fn = os.path.join(dirpath, filename)
partial_name = fn[len(path) + 1:]
for matcher in self.matchers:
if matcher.matches(partial_name):
results.add(File(partial_name, matcher.tags))
break
return results
def submitJobs(self, jobname, files):
self.client.waitForServer(90)
ret = []
for f in files:
output = self.makeOutput(f)
output = json.dumps(output).encode('utf8')
job = gear.TextJob(jobname, output)
self.client.submitJob(job, background=True)
ret.append(dict(handle=job.handle,
arguments=output))
return ret
def makeOutput(self, file_object):
output = {}
output['retry'] = False
output['event'] = self.makeEvent(file_object)
output['source_url'] = output['event']['fields']['log_url']
return output
def makeEvent(self, file_object):
out_event = {}
out_event["fields"] = self.makeFields(file_object.name)
basename = os.path.basename(file_object.name)
out_event["tags"] = [basename] + file_object.tags
if basename.endswith(".gz"):
# Backward compat for e-r which relies on tag values
# without the .gx suffix
out_event["tags"].append(basename[:-3])
return out_event
def makeFields(self, filename):
hosts = [h for h in self.hosts.values() if 'nodepool' in h]
zuul = self.zuul
fields = {}
fields["filename"] = filename
fields["build_name"] = zuul['job']
fields["build_status"] = self.success and 'SUCCESS' or 'FAILURE'
# TODO: this is too simplistic for zuul v3 multinode jobs
node = hosts[0]
fields["build_node"] = node['nodepool']['label']
fields["build_hostids"] = [h['nodepool']['host_id'] for h in hosts
if 'host_id' in h['nodepool']]
# TODO: should be build_executor, or removed completely
fields["build_master"] = zuul['executor']['hostname']
fields["project"] = zuul['project']['name']
# The voting value is "1" for voting, "0" for non-voting
fields["voting"] = int(zuul['voting'])
# TODO(clarkb) can we do better without duplicated data here?
fields["build_uuid"] = zuul['build']
fields["build_short_uuid"] = fields["build_uuid"][:7]
# TODO: this should be build_pipeline
fields["build_queue"] = zuul['pipeline']
# TODO: this is not interesteding anymore
fields["build_ref"] = zuul['ref']
fields["build_branch"] = zuul.get('branch', 'UNKNOWN')
# TODO: remove
fields["build_zuul_url"] = "N/A"
if 'change' in zuul:
fields["build_change"] = zuul['change']
fields["build_patchset"] = zuul['patchset']
elif 'newrev' in zuul:
fields["build_newrev"] = zuul.get('newrev', 'UNKNOWN')
fields["node_provider"] = node['nodepool']['provider']
log_url = urllib.parse.urljoin(self.log_url, filename)
fields["log_url"] = log_url
if 'executor' in zuul and 'hostname' in zuul['executor']:
fields["zuul_executor"] = zuul['executor']['hostname']
if 'attempts' in zuul:
fields["zuul_attempts"] = zuul['attempts']
return fields
def main():
module = AnsibleModule(
argument_spec=dict(
gearman_server=dict(type='str'),
gearman_port=dict(type='int', default=4730),
# TODO: add ssl support
host_vars=dict(type='dict'),
path=dict(type='path'),
config=dict(type='dict'),
success=dict(type='bool'),
log_url=dict(type='str'),
job=dict(type='str'),
),
)
p = module.params
results = dict(files=[], jobs=[], invocation={})
try:
l = LogMatcher(p.get('gearman_server'),
p.get('gearman_port'),
p.get('config'),
p.get('success'),
p.get('log_url'),
p.get('host_vars'))
files = l.findFiles(p['path'])
for f in files:
results['files'].append(f.toDict())
for handle in l.submitJobs(p['job'], files):
results['jobs'].append(handle)
module.exit_json(**results)
except Exception:
tb = traceback.format_exc()
module.fail_json(msg='Unknown error',
details=tb,
**results)
if __name__ == '__main__':
main()

3526
roles/submit-log-processor-jobs/module_utils/gear.py
View File

File diff suppressed because it is too large Load Diff

289
roles/submit-log-processor-jobs/module_utils/gear_acl.py
View File

@ -1,289 +0,0 @@
# Copyright 2014 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import re
class ACLError(Exception):
pass
class ACLEntry(object):
"""An access control list entry.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
:arg str register: A regular expression that matches the jobs that
connections with this certificate are permitted to register.
:arg str invoke: A regular expression that matches the jobs that
connections with this certificate are permitted to invoke.
Also implies the permission to cancel the same set of jobs in
the queue.
:arg boolean grant: A flag indicating whether connections with
this certificate are permitted to grant access to other
connections. Also implies the permission to revoke access
from other connections. The ability to self-revoke access is
always implied.
"""
def __init__(self, subject, register=None, invoke=None, grant=False):
self.subject = subject
self.setRegister(register)
self.setInvoke(invoke)
self.setGrant(grant)
def __repr__(self):
return ('<ACLEntry for %s register=%s invoke=%s grant=%s>' %
(self.subject, self.register, self.invoke, self.grant))
def isEmpty(self):
"""Checks whether this entry grants any permissions at all.
:returns: False if any permission is granted, otherwise True.
"""
if (self.register is None and
self.invoke is None and
self.grant is False):
return True
return False
def canRegister(self, name):
"""Check whether this subject is permitted to register a function.
:arg str name: The function name to check.
:returns: A boolean indicating whether the action should be permitted.
"""
if self.register is None:
return False
if not self._register.match(name):
return False
return True
def canInvoke(self, name):
"""Check whether this subject is permitted to register a function.
:arg str name: The function name to check.
:returns: A boolean indicating whether the action should be permitted.
"""
if self.invoke is None:
return False
if not self._invoke.match(name):
return False
return True
def setRegister(self, register):
"""Sets the functions that this subject can register.
:arg str register: A regular expression that matches the jobs that
connections with this certificate are permitted to register.
"""
self.register = register
if register:
try:
self._register = re.compile(register)
except re.error as e:
raise ACLError('Regular expression error: %s' % (e.message,))
else:
self._register = None
def setInvoke(self, invoke):
"""Sets the functions that this subject can invoke.
:arg str invoke: A regular expression that matches the jobs that
connections with this certificate are permitted to invoke.
"""
self.invoke = invoke
if invoke:
try:
self._invoke = re.compile(invoke)
except re.error as e:
raise ACLError('Regular expression error: %s' % (e.message,))
else:
self._invoke = None
def setGrant(self, grant):
"""Sets whether this subject can grant ACLs to others.
:arg boolean grant: A flag indicating whether connections with
this certificate are permitted to grant access to other
connections. Also implies the permission to revoke access
from other connections. The ability to self-revoke access is
always implied.
"""
self.grant = grant
class ACL(object):
"""An access control list.
ACLs are deny-by-default. The checked actions are only allowed if
there is an explicit rule in the ACL granting permission for a
given client (identified by SSL certificate Common Name Subject)
to perform that action.
"""
def __init__(self):
self.subjects = {}
def add(self, entry):
"""Add an ACL entry.
:arg Entry entry: The :py:class:`ACLEntry` to add.
:raises ACLError: If there is already an entry for the subject.
"""
if entry.subject in self.subjects:
raise ACLError("An ACL entry for %s already exists" %
(entry.subject,))
self.subjects[entry.subject] = entry
def remove(self, subject):
"""Remove an ACL entry.
:arg str subject: The SSL certificate Subject Common Name to
remove from the ACL.
:raises ACLError: If there is no entry for the subject.
"""
if subject not in self.subjects:
raise ACLError("There is no ACL entry for %s" % (subject,))
del self.subjects[subject]
def getEntries(self):
"""Return a list of current ACL entries.
:returns: A list of :py:class:`ACLEntry` objects.
"""
items = list(self.subjects.items())
items.sort(key=lambda a: a[0])
return [x[1] for x in items]
def canRegister(self, subject, name):
"""Check whether a subject is permitted to register a function.
:arg str subject: The SSL certificate Subject Common Name to
check against.
:arg str name: The function name to check.
:returns: A boolean indicating whether the action should be permitted.
"""
entry = self.subjects.get(subject)
if entry is None:
return False
return entry.canRegister(name)
def canInvoke(self, subject, name):
"""Check whether a subject is permitted to invoke a function.
:arg str subject: The SSL certificate Subject Common Name to
check against.
:arg str name: The function name to check.
:returns: A boolean indicating whether the action should be permitted.
"""
entry = self.subjects.get(subject)
if entry is None:
return False
return entry.canInvoke(name)
def canGrant(self, subject):
"""Check whether a subject is permitted to grant access to others.
:arg str subject: The SSL certificate Subject Common Name to
check against.
:returns: A boolean indicating whether the action should be permitted.
"""
entry = self.subjects.get(subject)
if entry is None:
return False
if not entry.grant:
return False
return True
def grantInvoke(self, subject, invoke):
"""Grant permission to invoke certain functions.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
:arg str invoke: A regular expression that matches the jobs
that connections with this certificate are permitted to
invoke. Also implies the permission to cancel the same
set of jobs in the queue.
"""
e = self.subjects.get(subject)
if not e:
e = ACLEntry(subject)
self.add(e)
e.setInvoke(invoke)
def grantRegister(self, subject, register):
"""Grant permission to register certain functions.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
:arg str register: A regular expression that matches the jobs that
connections with this certificate are permitted to register.
"""
e = self.subjects.get(subject)
if not e:
e = ACLEntry(subject)
self.add(e)
e.setRegister(register)
def grantGrant(self, subject):
"""Grant permission to grant permissions to other connections.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
"""
e = self.subjects.get(subject)
if not e:
e = ACLEntry(subject)
self.add(e)
e.setGrant(True)
def revokeInvoke(self, subject):
"""Revoke permission to invoke all functions.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
"""
e = self.subjects.get(subject)
if e:
e.setInvoke(None)
if e.isEmpty():
self.remove(subject)
def revokeRegister(self, subject):
"""Revoke permission to register all functions.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
"""
e = self.subjects.get(subject)
if e:
e.setRegister(None)
if e.isEmpty():
self.remove(subject)
def revokeGrant(self, subject):
"""Revoke permission to grant permissions to other connections.
:arg str subject: The SSL certificate Subject Common Name to which
the entry applies.
"""
e = self.subjects.get(subject)
if e:
e.setGrant(False)
if e.isEmpty():
self.remove(subject)

83
roles/submit-log-processor-jobs/module_utils/gear_constants.py
View File

@ -1,83 +0,0 @@
# Copyright 2013 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Protocol Constants
==================
These are not necessary for normal API usage. See the `Gearman
protocol reference <http://gearman.org/protocol>`_ for an explanation
of each of these.
Magic Codes
-----------
.. py:data:: REQ
The Gearman magic code for a request.
.. py:data:: RES
The Gearman magic code for a response.
Packet Types
------------
"""
types = {
1: 'CAN_DO',
2: 'CANT_DO',
3: 'RESET_ABILITIES',
4: 'PRE_SLEEP',
# unused
6: 'NOOP',
7: 'SUBMIT_JOB',
8: 'JOB_CREATED',
9: 'GRAB_JOB',
10: 'NO_JOB',
11: 'JOB_ASSIGN',
12: 'WORK_STATUS',
13: 'WORK_COMPLETE',
14: 'WORK_FAIL',
15: 'GET_STATUS',
16: 'ECHO_REQ',
17: 'ECHO_RES',
18: 'SUBMIT_JOB_BG',
19: 'ERROR',
20: 'STATUS_RES',
21: 'SUBMIT_JOB_HIGH',
22: 'SET_CLIENT_ID',
23: 'CAN_DO_TIMEOUT',
24: 'ALL_YOURS',
25: 'WORK_EXCEPTION',
26: 'OPTION_REQ',
27: 'OPTION_RES',
28: 'WORK_DATA',
29: 'WORK_WARNING',
30: 'GRAB_JOB_UNIQ',
31: 'JOB_ASSIGN_UNIQ',
32: 'SUBMIT_JOB_HIGH_BG',
33: 'SUBMIT_JOB_LOW',
34: 'SUBMIT_JOB_LOW_BG',
35: 'SUBMIT_JOB_SCHED',
36: 'SUBMIT_JOB_EPOCH',
}
for i, name in types.items():
globals()[name] = i
__doc__ += '\n.. py:data:: %s\n' % name
REQ = b'\x00REQ'
RES = b'\x00RES'

44
roles/submit-logstash-jobs/README.rst
View File

@ -1,44 +0,0 @@
Submit a log processing job to the logstash workers.
This role examines all of the files in the log subdirectory of the job
work dir and any matching filenames are submitted to the gearman queue
for the logstash log processor, along with any tags configured for
those filenames.
**Role Variables**
.. zuul:rolevar:: logstash_gearman_server
:default: logstash.openstack.org
The gearman server to use.
.. zuul:rolevar:: logstash_processor_config
:type: dict
The default file configuration for the logstash parser.
This is a dictionary that contains a single entry:
.. zuul:rolevar:: files
:type: list
A list of files to search for in the ``work/logs/`` directory on
the executor. Each file will be compared to the entries in this
list, and if it matches, a processing job will be submitted to
the logstash processing queue, along with the tags for the
matching entry. Order is important: the first matcing is used.
This field is list of dictionaries, as follows:
.. zuul:rolevar:: name
The name of the file to process. This is treated as an
unanchored regular expression. To match the full path
(underneath ``work/logs``) start and end the string with
``^`` and ``$`` respectively.
.. zuul:rolevar:: tags
:type: list
A list of strings indicating the logstash processing tags
associated with this file. These may be used to indicate the
file format to the parser.

88
roles/submit-logstash-jobs/defaults/main.yaml
View File

@ -1,88 +0,0 @@
logstash_gearman_server: logstash.openstack.org
# For every file found in the logs directory (and its subdirs), the
# module will attempt to match the filenames below. If there is a
# match, the file is submitted to the logstash processing queue, along
# with the tags for that match. The first match wins, so be sure to
# list more specific names first. The names are un-anchored regular
# expressions (so if you need to match the root (i.e, the work/logs/
# directory), be sure to anchor them with ^).
logstash_processor_config:
files:
- name: job-output\.txt
tags:
- console
- console.html
- name: grenade\.sh\.txt
tags:
- console
- console.html
- name: devstacklog\.txt(?!.*summary)
tags:
- console
- console.html
- name: apache/keystone\.txt
tags:
- screen
- oslofmt
- name: apache/horizon_error\.txt
tags:
- apacheerror
# TODO(clarkb) Add swift proxy logs here.
- name: syslog\.txt
tags:
- syslog
- name: tempest\.txt
tags:
- screen
- oslofmt
- name: javelin\.txt
tags:
- screen
- oslofmt
# Neutron index log files (files with messages from all test cases)
- name: dsvm-functional-index\.txt
tags:
- oslofmt
- name: dsvm-fullstack-index\.txt
tags:
- oslofmt
- name: screen-s-account\.txt
tags:
- screen
- apachecombined
- name: screen-s-container\.txt
tags:
- screen
- apachecombined
- name: screen-s-object\.txt
tags:
- screen
- apachecombined
# tripleo logs
- name: postci\.txt
tags:
- console
- postci
- name: var/log/extra/logstash\.txt
tags:
- console
- postci
- name: var/log/extra/errors\.txt
tags:
- console
- errors
# wildcard logs
- name: devstack-gate-.*\.txt
tags:
- console
- console.html
# NOTE(mriedem): Logs that are known logstash index OOM killers are
# blacklisted here until fixed.
# screen-kubelet.txt: https://bugs.launchpad.net/kuryr-kubernetes/+bug/1795067
# screen-mistral-engine.txt: https://bugs.launchpad.net/mistral/+bug/1795068
# screen-monasca-persister.txt: https://storyboard.openstack.org/#!/story/2003911
# screen-ovn-northd.txt: https://bugs.launchpad.net/networking-ovn/+bug/1795069
- name: screen-(?!(peakmem_tracker|dstat|karaf|kubelet|mistral-engine|monasca-persister|monasca-api|ovn-northd|q-svc)).*\.txt
tags:
- screen
- oslofmt

2
roles/submit-logstash-jobs/meta/main.yaml
View File

@ -1,2 +0,0 @@
dependencies:
- role: submit-log-processor-jobs

9
roles/submit-logstash-jobs/tasks/main.yaml
View File

@ -1,9 +0,0 @@
- name: Submit logstash processing jobs to log processors
submit_log_processor_jobs:
gearman_server: "{{ logstash_gearman_server }}"
job: "push-log"
config: "{{ logstash_processor_config }}"
success: "{{ zuul_success }}"
host_vars: "{{ hostvars }}"
path: "{{ zuul.executor.log_root }}"
log_url: "{{ (lookup('file', zuul.executor.result_data_file) | from_json).get('data').get('zuul').get('log_url') }}"

36
roles/submit-subunit-jobs/README.rst
View File

@ -1,36 +0,0 @@
Submit a log processing job to the subunit workers.
This role examines all of the files in the log subdirectory of the job
work dir and any matching filenames are submitted to the gearman queue
for the subunit log processor.
**Role Variables**
.. zuul:rolevar:: subunit_gearman_server
:default: logstash.openstack.org
The gearman server to use.
.. zuul:rolevar:: subunit_processor_config
:type: dict
The default file configuration for the subunit parser.
This is a dictionary that contains a single entry:
.. zuul:rolevar:: files
:type: list
A list of files to search for in the ``work/logs/`` directory on
the executor. Each file will be compared to the entries in this
list, and if it matches, a processing job will be submitted to
the subunit processing queue, along with the tags for the
matching entry. Order is important: the first matcing is used.
This field is list of dictionaries, as follows:
.. zuul:rolevar:: name
The name of the file to process. This is treated as an
unanchored regular expression. To match the full path
(underneath ``work/logs``) start and end the string with
``^`` and ``$`` respectively.

12
roles/submit-subunit-jobs/defaults/main.yaml
View File

@ -1,12 +0,0 @@
subunit_gearman_server: logstash.openstack.org
# For every file found in the logs directory (and its subdirs), the
# module will attempt to match the filenames below. If there is a
# match, the file is submitted to the subunit processing queue, along
# with the tags for that match. The first match wins, so be sure to
# list more specific names first. The names are un-anchored regular
# expressions (so if you need to match the root (i.e, the work/logs/
# directory), be sure to anchor them with ^).
subunit_processor_config:
files:
- name: testrepository.subunit
- name: karma.subunit

2
roles/submit-subunit-jobs/meta/main.yaml
View File

@ -1,2 +0,0 @@
dependencies:
- role: submit-log-processor-jobs

10
roles/submit-subunit-jobs/tasks/main.yaml
View File

@ -1,10 +0,0 @@
- name: Submit subunit processing jobs to log processors
when: zuul.pipeline in ['gate', 'periodic', 'post']
submit_log_processor_jobs:
gearman_server: "{{ subunit_gearman_server }}"
job: "push-subunit"
config: "{{ subunit_processor_config }}"
success: "{{ zuul_success }}"
host_vars: "{{ hostvars }}"
path: "{{ zuul.executor.log_root }}"
log_url: "{{ (lookup('file', zuul.executor.result_data_file) | from_json).get('zuul').get('log_url') }}"