jasima/majuna
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

metrics: remove type ignores that are no longer required

Browse source
This commit is contained in:
Iain Learmonth 2022-11-02 13:58:54 +00:00
parent 0ff61721d9
commit 0ab9fdebc5
2 changed files with 14 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

8
app/__init__.py
View file

@ -16,7 +16,7 @@ app = Flask(__name__)
app.config.from_file("../config.yaml", load=yaml.safe_load) app.config.from_file("../config.yaml", load=yaml.safe_load)
app.wsgi_app = DispatcherMiddleware(app.wsgi_app, { # type: ignore[assignment] app.wsgi_app = DispatcherMiddleware(app.wsgi_app, { # type: ignore[assignment]
'/metrics': make_wsgi_app() # type: ignore[no-untyped-call] '/metrics': make_wsgi_app()
}) })
db.init_app(app) db.init_app(app)
@ -30,9 +30,9 @@ app.register_blueprint(tfstate, url_prefix="/tfstate")
if 'nose' not in sys.modules.keys() and sys.argv[1] != "db": if 'nose' not in sys.modules.keys() and sys.argv[1] != "db":
from app.metrics import DefinedProxiesCollector, BlockedProxiesCollector, AutomationCollector from app.metrics import DefinedProxiesCollector, BlockedProxiesCollector, AutomationCollector
REGISTRY.register(DefinedProxiesCollector(app)) # type: ignore[no-untyped-call] REGISTRY.register(DefinedProxiesCollector(app))
REGISTRY.register(BlockedProxiesCollector(app)) # type: ignore[no-untyped-call] REGISTRY.register(BlockedProxiesCollector(app))
REGISTRY.register(AutomationCollector(app)) # type: ignore[no-untyped-call] REGISTRY.register(AutomationCollector(app))
@app.route('/') @app.route('/')

19
app/metrics.py
View file

@ -3,13 +3,14 @@ from typing import Iterator, Optional
from flask import Flask from flask import Flask
from prometheus_client.metrics_core import GaugeMetricFamily, CounterMetricFamily, Metric from prometheus_client.metrics_core import GaugeMetricFamily, CounterMetricFamily, Metric
from prometheus_client.registry import Collector
from sqlalchemy import text from sqlalchemy import text
from app.extensions import db from app.extensions import db
from app.models.automation import Automation, AutomationState from app.models.automation import Automation, AutomationState
class FlaskCollector: class FlaskCollector(Collector):
_app: Optional[Flask] _app: Optional[Flask]
def __init__(self, app: Optional[Flask] = None) -> None: def __init__(self, app: Optional[Flask] = None) -> None:
@ -43,10 +44,10 @@ class DefinedProxiesCollector(FlaskCollector):
""")) """))
c = GaugeMetricFamily("defined_proxies", "Number of proxies currently defined for deployment", c = GaugeMetricFamily("defined_proxies", "Number of proxies currently defined for deployment",
labels=['group_id', 'group_name', 'provider', 'pool_id', labels=['group_id', 'group_name', 'provider', 'pool_id',
'pool_name']) # type: ignore[no-untyped-call] 'pool_name'])
for row in result: for row in result:
c.add_metric([str(row[0]), row[1], row[2], str(row[3]), row[4]], c.add_metric([str(row[0]), row[1], row[2], str(row[3]), row[4]],
row[5]) # type: ignore[no-untyped-call] row[5])
yield c yield c
@ -66,25 +67,25 @@ class BlockedProxiesCollector(FlaskCollector):
c = CounterMetricFamily("deprecated_proxies", c = CounterMetricFamily("deprecated_proxies",
"Number of proxies deprecated", "Number of proxies deprecated",
labels=['group_id', 'group_name', 'provider', 'pool_id', 'pool_name', labels=['group_id', 'group_name', 'provider', 'pool_id', 'pool_name',
'deprecation_reason']) # type: ignore[no-untyped-call] 'deprecation_reason'])
for row in result: for row in result:
c.add_metric([str(row[0]), row[1], row[2], str(row[3]), row[4], row[5]], c.add_metric([str(row[0]), row[1], row[2], str(row[3]), row[4], row[5]],
row[6]) # type: ignore[no-untyped-call] row[6])
yield c yield c
class AutomationCollector(FlaskCollector): class AutomationCollector(FlaskCollector):
def collect_in_ctx(self, app: Flask) -> Iterator[Metric]: def collect_in_ctx(self, app: Flask) -> Iterator[Metric]:
c = GaugeMetricFamily("automation_state", "The automation state (0: idle, 1: running, 2: error)", c = GaugeMetricFamily("automation_state", "The automation state (0: idle, 1: running, 2: error)",
labels=['automation_name']) # type: ignore[no-untyped-call] labels=['automation_name'])
automations = Automation.query.all() automations = Automation.query.all()
for automation in automations: for automation in automations:
if automation.short_name in app.config['HIDDEN_AUTOMATIONS']: if automation.short_name in app.config['HIDDEN_AUTOMATIONS']:
continue continue
if automation.state == AutomationState.IDLE: if automation.state == AutomationState.IDLE:
c.add_metric([automation.short_name], 0) # type: ignore[no-untyped-call] c.add_metric([automation.short_name], 0)
elif automation.state == AutomationState.RUNNING: elif automation.state == AutomationState.RUNNING:
c.add_metric([automation.short_name], 1) # type: ignore[no-untyped-call] c.add_metric([automation.short_name], 1)
else: else:
c.add_metric([automation.short_name], 2) # type: ignore[no-untyped-call] c.add_metric([automation.short_name], 2)
yield c yield c