File indexing completed on 2026-11-09 10:47:34
0001 """OSG throttle incidents: a pool our OSG pilots run in stops giving them
0002 new slots for a limit on its own side (docs/OSG_SUBMIT_REPORTER.md,
0003 Fair-share limit).
0004
0005 The submit host reporter's record carries each pool negotiator's last
0006 cycle and whether it held our submitter at its fair-share limit. On each
0007 record this compares the reading with the state kept in PersistentState
0008 and records one action-stream incident per change: ``osg_throttle`` at
0009 info level when a pool first holds us (OSG's normal mode of operation,
0010 never a warning: Torre Wenaus, 2026-10-28), and its lift once the pool has
0011 released us for LIFT_READINGS consecutive records, so a limit that flips
0012 between negotiation cycles is one incident rather than one per cycle.
0013 The epicprod-live subscription and a capcom subscription deliver it
0014 (docs/NOTICE_ROUTING.md).
0015 """
0016 import logging
0017
0018 from monitor_app.short_names import short_name
0019
0020 logger = logging.getLogger(__name__)
0021
0022 ACTION = 'osg_throttle'
0023 STATE_KEY = 'osg_throttle'
0024 LIFT_READINGS = 3
0025
0026
0027 def readings(record):
0028 """{pool: {'held', 'submitters', 'cycle_at'}} for the pools whose
0029 negotiators the record read, by short pool name; None when the record
0030 carries no negotiator reading."""
0031 negs = (record or {}).get('negotiators')
0032 if not isinstance(negs, dict) or not negs.get('negotiators'):
0033 return None
0034 out = {}
0035 for n in negs['negotiators']:
0036 pool = short_name(n.get('pool'))
0037 entry = out.setdefault(pool, {'held': False, 'submitters': [], 'cycle_at': None})
0038 if n.get('share_limited'):
0039 entry['held'] = True
0040 entry['submitters'] = sorted(set(entry['submitters']) | set(n['share_limited']))
0041 entry['cycle_at'] = max(filter(None, (entry['cycle_at'], n.get('cycle_at'))),
0042 default=None)
0043 return out
0044
0045
0046 def observe(record, username=''):
0047 """Record the incidents this record's reading implies and keep the
0048 state they are judged against."""
0049 from monitor_app.epicprod_logging import log_epicprod_action
0050 from monitor_app.models import PersistentState
0051
0052 read = readings(record)
0053 if read is None:
0054 return
0055 state = dict(PersistentState.get_state().get(STATE_KEY) or {})
0056 changed = False
0057 for pool, now in sorted(read.items()):
0058 before = state.get(pool) or {'state': 'clear', 'clear_readings': 0}
0059 if now['held']:
0060 if before['state'] != 'held':
0061 who = ', '.join(now['submitters'])
0062 text = (f'{pool} negotiator holds our submitter {who} at its fair-share '
0063 f'limit: no new OSG slots in {pool} until our usage share falls; '
0064 f'running pilots keep theirs.')
0065 log_epicprod_action(
0066 'report', ACTION, subject_type='osg_pool', subject_key=pool,
0067 subject_label=pool, username=username, outcome='held',
0068 sublevel='high', live_default=True,
0069 message=text, summary=text, severity='info', state='held',
0070 kind='fair_share', submitters=now['submitters'],
0071 cycle_at=now['cycle_at'])
0072 state[pool] = {'state': 'held', 'since': now['cycle_at'], 'clear_readings': 0}
0073 changed = True
0074 elif before.get('clear_readings'):
0075 state[pool] = dict(before, clear_readings=0)
0076 changed = True
0077 elif before['state'] == 'held':
0078 count = int(before.get('clear_readings') or 0) + 1
0079 if count >= LIFT_READINGS:
0080 text = (f'{pool} negotiator released our submitter from its fair-share '
0081 f'limit (held since {before.get("since") or "unknown"}, clear for '
0082 f'{count} consecutive readings).')
0083 log_epicprod_action(
0084 'report', ACTION, subject_type='osg_pool', subject_key=pool,
0085 subject_label=pool, username=username, outcome='ok',
0086 sublevel='high', live_default=True, message=text, summary=text,
0087 severity='info', state='lifted', kind='fair_share',
0088 held_since=before.get('since'))
0089 state[pool] = {'state': 'clear', 'clear_readings': 0}
0090 else:
0091 state[pool] = dict(before, clear_readings=count)
0092 changed = True
0093 if changed:
0094 PersistentState.update_state({STATE_KEY: state})
