Closes #18362: Create a system job for census reporting

This commit is contained in:
Jeremy Stretch 2025-01-09 09:53:28 -05:00
parent 80e1fd02bb
commit b11f179527
2 changed files with 46 additions and 15 deletions

View File

@ -1,8 +1,11 @@
import logging import logging
import requests
import sys
from netbox.jobs import JobRunner from django.conf import settings
from netbox.jobs import JobRunner, system_job
from netbox.search.backends import search_backend from netbox.search.backends import search_backend
from .choices import DataSourceStatusChoices from .choices import DataSourceStatusChoices, JobIntervalChoices
from .exceptions import SyncError from .exceptions import SyncError
from .models import DataSource from .models import DataSource
@ -31,3 +34,44 @@ class SyncDataSourceJob(JobRunner):
if type(e) is SyncError: if type(e) is SyncError:
logging.error(e) logging.error(e)
raise e raise e
@system_job(interval=JobIntervalChoices.INTERVAL_DAILY)
class SystemHousekeepingJob(JobRunner):
"""
Perform daily system housekeeping functions.
"""
class Meta:
name = "System Housekeeping"
def run(self, *args, **kwargs):
# Skip if running in development or test mode
if settings.DEBUG or 'test' in sys.argv:
return
# TODO: Migrate other housekeeping functions from the `housekeeping` management command.
self.send_census_report()
@staticmethod
def send_census_report():
"""
Send a census report (if enabled).
"""
# Skip if census reporting is disabled
if settings.ISOLATED_DEPLOYMENT or not settings.CENSUS_REPORTING_ENABLED:
return
census_data = {
'version': settings.RELEASE.full_version,
'python_version': sys.version.split()[0],
'deployment_id': settings.DEPLOYMENT_ID,
}
try:
requests.get(
url=settings.CENSUS_URL,
params=census_data,
timeout=3,
proxies=settings.HTTP_PROXIES
)
except requests.exceptions.RequestException:
pass

View File

@ -5,9 +5,7 @@ import os
import platform import platform
import sys import sys
import warnings import warnings
from urllib.parse import urlencode
import requests
from django.contrib.messages import constants as messages from django.contrib.messages import constants as messages
from django.core.exceptions import ImproperlyConfigured, ValidationError from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.core.validators import URLValidator from django.core.validators import URLValidator
@ -583,17 +581,6 @@ if SENTRY_ENABLED:
# Calculate a unique deployment ID from the secret key # Calculate a unique deployment ID from the secret key
DEPLOYMENT_ID = hashlib.sha256(SECRET_KEY.encode('utf-8')).hexdigest()[:16] DEPLOYMENT_ID = hashlib.sha256(SECRET_KEY.encode('utf-8')).hexdigest()[:16]
CENSUS_URL = 'https://census.netbox.oss.netboxlabs.com/api/v1/' CENSUS_URL = 'https://census.netbox.oss.netboxlabs.com/api/v1/'
CENSUS_PARAMS = {
'version': RELEASE.full_version,
'python_version': sys.version.split()[0],
'deployment_id': DEPLOYMENT_ID,
}
if CENSUS_REPORTING_ENABLED and not ISOLATED_DEPLOYMENT and not DEBUG and 'test' not in sys.argv:
try:
# Report anonymous census data
requests.get(f'{CENSUS_URL}?{urlencode(CENSUS_PARAMS)}', timeout=3, proxies=HTTP_PROXIES)
except requests.exceptions.RequestException:
pass
# #