# Copyright (c) 2017 LINE Corporation
# These sources are released under the terms of the MIT license: see LICENSE
import json
import logging
import django.contrib.sites.models
from django.conf import settings
from django.contrib.auth.models import Group, User
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.db import models, transaction
from django.forms.models import model_to_dict
from django.urls import reverse
from django.utils import timezone
from django.utils.functional import cached_property
from django.utils.text import slugify
import promgen.templatetags.promgen as macro
from promgen import plugins, tests, util, validators
from promgen.shortcuts import resolve_domain
logger = logging.getLogger(__name__)
[docs]
class Site(django.contrib.sites.models.Site):
# Proxy model for sites so that we can easily
# query our related Rules
rule_set = GenericRelation("promgen.Rule", for_concrete_model=False)
def get_absolute_url(self):
return reverse("site-detail")
class Meta:
proxy = True
[docs]
class Group(Group):
def get_absolute_url(self):
return reverse("group-detail", kwargs={"pk": self.pk})
class Meta:
proxy = True
ordering = ["name"]
class ObjectFilterManager(models.Manager):
def create(self, *args, **kwargs):
if "obj" in kwargs:
obj = kwargs.pop("obj")
kwargs["object_id"] = obj.id
kwargs["content_type_id"] = ContentType.objects.get_for_model(obj).id
return self.get_queryset().create(*args, **kwargs)
def filter(self, *args, **kwargs):
if "obj" in kwargs:
obj = kwargs.pop("obj")
kwargs["object_id"] = obj.id
kwargs["content_type_id"] = ContentType.objects.get_for_model(obj).id
return self.get_queryset().filter(*args, **kwargs)
def get_or_create(self, *args, **kwargs):
if "obj" in kwargs:
obj = kwargs.pop("obj")
kwargs["object_id"] = obj.id
kwargs["content_type_id"] = ContentType.objects.get_for_model(obj).id
if "defaults" in kwargs and "obj" in kwargs["defaults"]:
obj = kwargs["defaults"].pop("obj")
kwargs["defaults"]["object_id"] = obj.id
kwargs["defaults"]["content_type_id"] = ContentType.objects.get_for_model(obj).id
return self.get_queryset().get_or_create(*args, **kwargs)
[docs]
class Sender(models.Model):
objects = ObjectFilterManager()
sender = models.CharField(max_length=128)
value = models.CharField(max_length=128)
alias = models.CharField(max_length=128, blank=True)
content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
limit_choices_to=(
models.Q(app_label="auth", model="user")
| models.Q(app_label="promgen", model="project")
| models.Q(app_label="promgen", model="service")
),
)
object_id = models.PositiveIntegerField()
content_object = GenericForeignKey("content_type", "object_id")
owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
enabled = models.BooleanField(default=True)
def show_value(self):
if self.alias:
return self.alias
if self.sender == "promgen.notification.user":
try:
return User.objects.get(pk=self.value).username
except User.DoesNotExist:
return None
return self.value
show_value.short_description = "Value"
def __str__(self):
return f"{self.sender}:{self.show_value()}"
[docs]
@classmethod
def driver_set(cls):
"""Return the list of drivers for Sender model"""
for entry in plugins.notifications():
try:
yield entry.module_name, entry.load()
except ImportError:
logger.warning("Error importing %s", entry.module_name)
__driver = {}
@property
def driver(self):
"""Return configured driver for Sender model instance"""
if self.sender in self.__driver:
return self.__driver[self.sender]
for entry in plugins.notifications():
try:
self.__driver[entry.module_name] = entry.load()()
except ImportError:
logger.warning("Error importing %s", entry.module_name)
return self.__driver[self.sender]
[docs]
def test(self):
"""
Test sender plugin
Uses the same test json from our unittests but subs in the currently
tested object as part of the test data
"""
data = tests.Data("examples", "alertmanager.json").json()
if hasattr(self.content_object, "name"):
data["commonLabels"][self.content_type.name] = self.content_object.name
for alert in data.get("alerts", []):
alert["labels"][self.content_type.name] = self.content_object.name
from promgen import tasks
tasks.send_alert(self.sender, self.value, data)
[docs]
def filtered(self, alert):
"""
Check filters for a specific sender
If no filters are defined, then we let the message through
If filters are defined, then we check to see if at least one filter matches
If no filters match, then we assume it's filtered out
"""
logger.debug("Checking labels %s", alert["commonLabels"])
# If we have no specific whitelist, then we let everything through
if self.filter_set.count() == 0:
return False
# If we have filters defined, then we need to check to see if our
# filters match
for f in self.filter_set.all():
logger.debug("Checking filter %s %s", f.name, f.value)
if alert["commonLabels"].get(f.name) == f.value:
return False
# If none of our filters match, then we blacklist this sender
return True
def get_absolute_url(self):
return reverse("notifier-edit", kwargs={"pk": self.pk})
[docs]
class Filter(models.Model):
sender = models.ForeignKey("Sender", on_delete=models.CASCADE)
name = models.CharField(max_length=128)
value = models.CharField(max_length=128)
class Meta:
ordering = ("sender", "name", "value")
unique_together = (("sender", "name", "value"),)
[docs]
class Shard(models.Model):
name = models.CharField(max_length=128, unique=True, validators=[validators.labelvalue])
url = models.URLField(max_length=256)
authorization = models.CharField(
max_length=4083, # 4KB - len("authorization")
blank=True,
null=True,
help_text="HTTP Authorization header for this shard's API",
)
proxy = models.BooleanField(
default=False,
help_text="Queries can be proxied to these shards",
)
enabled = models.BooleanField(
default=True,
help_text="Able to register new Services and Projects",
)
samples = models.PositiveBigIntegerField(
default=5000000,
help_text="Estimated Sample Count",
)
targets = models.PositiveBigIntegerField(
default=10000,
help_text="Estimated Target Count",
)
class Meta:
ordering = ["name"]
def get_absolute_url(self):
return reverse("datasource-detail", kwargs={"pk": self.pk})
def __str__(self):
if self.enabled:
return self.name
return self.name + " (disabled)"
@property
def samples_count(self):
return self._query_count("sum(scrape_samples_scraped)", "sample count")
@property
def targets_count(self):
return self._query_count("count(up)", "target count")
def _query_count(self, query, metric_name):
try:
headers = {"Authorization": self.authorization} if self.authorization else {}
response = util.get(
f"{self.url}/api/v1/query", params={"query": query}, headers=headers
)
response.raise_for_status()
data = response.json()
value = data["data"]["result"][0]["value"][1]
return int(value)
except Exception as e:
logger.warning("Error fetching %s for shard %s: %s", metric_name, self.name, e)
return 0
[docs]
class Service(models.Model):
name = models.CharField(max_length=128, unique=True, validators=[validators.labelvalue])
description = models.TextField(blank=True)
owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
notifiers = GenericRelation(Sender)
rule_set = GenericRelation("Rule")
class Meta:
ordering = ["name"]
permissions = [
("service_admin", "Admin"),
("service_editor", "Editor"),
("service_viewer", "Viewer"),
]
def get_absolute_url(self):
return reverse("service-detail", kwargs={"pk": self.pk})
def __str__(self):
return self.name
@classmethod
def default(cls, service_name="Default", shard_name="Default"):
shard, created = Shard.objects.get_or_create(name=shard_name)
if created:
logger.info("Created default shard")
service, created = cls.objects.get_or_create(name=service_name, defaults={"shard": shard})
if created:
logger.info("Created default service")
return service
[docs]
class Project(models.Model):
name = models.CharField(max_length=128, unique=True, validators=[validators.labelvalue])
description = models.TextField(blank=True)
owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
service = models.ForeignKey("promgen.Service", on_delete=models.CASCADE)
shard = models.ForeignKey("promgen.Shard", on_delete=models.CASCADE)
notifiers = GenericRelation(Sender)
rule_set = GenericRelation("Rule")
class Meta:
ordering = ["name"]
permissions = [
("project_admin", "Admin"),
("project_editor", "Editor"),
("project_viewer", "Viewer"),
]
def get_absolute_url(self):
return reverse("project-detail", kwargs={"pk": self.pk})
def __str__(self):
return f"{self.service} » {self.name}"
[docs]
class Farm(models.Model):
name = models.CharField(max_length=128, validators=[validators.labelvalue])
source = models.CharField(max_length=128)
project = models.OneToOneField("promgen.Project", on_delete=models.CASCADE)
class Meta:
ordering = ["name"]
def get_absolute_url(self):
return reverse("farm-detail", kwargs={"pk": self.pk})
def refresh(self):
target = set()
current = {host.name for host in self.host_set.all()}
for entry in plugins.discovery():
if self.source == entry.name:
target.update(entry.load()().fetch(self.name))
remove = current - target
add = target - current
if add:
Audit.log(f"Adding {add} to {self}", self)
Host.objects.bulk_create([Host(name=name, farm_id=self.id) for name in add])
if remove:
Audit.log(f"Removing {add} from {self}", self)
Host.objects.filter(farm=self, name__in=remove).delete()
return add, remove
@classmethod
def fetch(cls, source):
for entry in plugins.discovery():
if entry.name == source:
yield from entry.load()().farms()
@cached_property
def driver(self):
"""Return configured driver for Farm model instance"""
for entry in plugins.discovery():
if entry.name == self.source:
return entry.load()()
@property
def editable(self):
return not self.driver.remote
[docs]
@classmethod
def driver_set(cls):
"""Return the list of drivers for Farm model"""
for entry in plugins.discovery():
yield entry.name, entry.load()()
def __str__(self):
return f"{self.name} ({self.source})"
[docs]
class Host(models.Model):
name = models.CharField(max_length=128)
farm = models.ForeignKey("Farm", on_delete=models.CASCADE)
class Meta:
ordering = ["name"]
unique_together = (("name", "farm"),)
def get_absolute_url(self):
return reverse("host-detail", kwargs={"slug": self.name})
def __str__(self):
return f"{self.name} [{self.farm.name}]"
[docs]
class BaseExporter(models.Model):
job = models.CharField(max_length=128, help_text="Exporter name. Example node, jmx, app")
port = models.IntegerField(help_text="Port Exporter is running on")
path = models.CharField(
max_length=128, blank=True, help_text="Exporter path. Defaults to /metrics"
)
scheme = models.CharField(
max_length=5,
choices=(("http", "http"), ("https", "https")),
default="http",
help_text="Scrape exporter over http or https",
)
class Meta:
abstract = True
@property
def path_value(self):
return self.path or "/metrics"
[docs]
class DefaultExporter(BaseExporter):
class Meta:
ordering = ["job", "port"]
unique_together = (("job", "port", "path"),)
[docs]
class Exporter(BaseExporter):
project = models.ForeignKey("Project", on_delete=models.CASCADE)
enabled = models.BooleanField(default=True)
class Meta:
ordering = ["job", "port"]
unique_together = (("job", "port", "path", "scheme", "project"),)
def __str__(self):
return "{}:{}{}".format(self.job, self.port, self.path_value)
[docs]
class Probe(models.Model):
module = models.CharField(
help_text="Probe Module from blackbox_exporter config", max_length=128, unique=True
)
description = models.TextField(blank=True)
class Meta:
ordering = ["module"]
def __str__(self):
return f"{self.module} » {self.description}"
[docs]
class URL(models.Model):
url = models.URLField(max_length=256, validators=[validators.scraped_url])
project = models.ForeignKey("Project", on_delete=models.CASCADE)
probe = models.ForeignKey("promgen.Probe", on_delete=models.CASCADE)
class Meta:
ordering = ["project__service", "project", "url"]
def __str__(self):
return f"{self.project} [{self.url}]"
[docs]
class Rule(models.Model):
objects = ObjectFilterManager()
name = models.CharField(max_length=128, unique=True, validators=[validators.metricname])
clause = models.TextField(help_text="Prometheus query")
duration = models.CharField(
max_length=128,
validators=[validators.duration],
help_text="Duration field with postfix. Example 30s, 5m, 1d",
)
enabled = models.BooleanField(default=True)
parent = models.ForeignKey(
"Rule", null=True, related_name="overrides", on_delete=models.SET_NULL
)
content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
limit_choices_to=(
models.Q(app_label="promgen", model="site")
| models.Q(app_label="promgen", model="project")
| models.Q(app_label="promgen", model="service")
),
)
object_id = models.PositiveIntegerField()
content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False)
description = models.TextField(blank=True)
labels = models.JSONField(default=dict)
annotations = models.JSONField(default=dict)
class Meta:
ordering = ["content_type", "object_id", "name"]
def __str__(self):
return (
f"{self.pk}:{self.name}"
if self.content_object is None
else f"{self.pk}:{self.name} [{self.content_object.name}]"
)
def get_absolute_url(self):
return reverse("rule-detail", kwargs={"pk": self.pk})
def set_object(self, content_type, object_id):
self.content_type = ContentType.objects.get(
model=content_type,
app_label="promgen",
)
self.object_id = object_id
[docs]
def copy_to(self, content_type, object_id):
"""
Make a copy under a new service
It's important that we set pk to None so a new object is created, but we
also need to ensure the new name is unique by appending some unique data
to the end of the name
"""
with transaction.atomic():
content_type = ContentType.objects.get(model=content_type, app_label="promgen")
# First check to see if this rule is already overwritten
for rule in Rule.objects.filter(
parent_id=self.pk, content_type=content_type, object_id=object_id
):
return rule
content_object = content_type.get_object_for_this_type(pk=object_id)
orig_pk = self.pk
self.pk = None
self.parent_id = orig_pk
self.name = f"{self.name}_{slugify(content_object.name)}".replace("-", "_")
self.content_type = content_type
self.object_id = object_id
# Enable the copy by default since it's more likely the user prefers
# to have their own copy enabled rather than the original one.
self.enabled = True
self.clause = self.clause.replace(
macro.EXCLUSION_MACRO,
f'{content_type.model}="{content_object.name}",{macro.EXCLUSION_MACRO}',
)
# Add a label to our new rule by default, to help ensure notifications
# get routed to the notifier we expect
self.labels[content_type.model] = content_object.name
self.save()
return self
# Custom logic before saving Rule to control the value of the annotation "rule":
# Format: annotations["rule"] = {domain}/rule/{id}
[docs]
def save(self, *args, **kwargs):
with transaction.atomic():
if self.pk:
# When updating rule, we already have the primary key.
# Just set annotations["rule"] before saving to database.
self.annotations["rule"] = resolve_domain("rule-detail", pk=self.pk)
super().save(*args, **kwargs)
else:
# When creating a new rule, the primary key is typically not available until the
# instance is saved to the database.
# Therefore, we save it first then set annotations["rule"] and save again.
super().save(*args, **kwargs)
self.annotations["rule"] = resolve_domain("rule-detail", pk=self.pk)
super().save(update_fields=["annotations"])
[docs]
class AlertLabel(models.Model):
alert = models.ForeignKey("Alert", on_delete=models.CASCADE)
name = models.CharField(max_length=128)
value = models.TextField()
[docs]
class Alert(models.Model):
created = models.DateTimeField(default=timezone.now)
body = models.TextField()
sent_count = models.PositiveIntegerField(default=0)
error_count = models.PositiveIntegerField(default=0)
def get_absolute_url(self):
return reverse("alert-detail", kwargs={"pk": self.pk})
def expand(self):
# Map of Prometheus labels to Promgen objects
LABEL_MAPPING = [
("project", Project),
("service", Service),
]
routable = {}
data = json.loads(self.body)
data.setdefault("commonLabels", {})
data.setdefault("commonAnnotations", {})
# Set our link back to Promgen for processed notifications
# The original externalURL can still be visible from the alerts page
data["externalURL"] = resolve_domain(self.get_absolute_url())
# Look through our labels and find the object from Promgen's DB
# If we find an object in Promgen, add an annotation with a direct link
for label, klass in LABEL_MAPPING:
if label not in data["commonLabels"]:
logger.debug("Missing label %s", label)
continue
# Should only find a single value, but I think filter is a little
# bit more forgiving than get in terms of throwing errors
for obj in klass.objects.filter(name=data["commonLabels"][label]):
logger.debug("Found %s %s", label, obj)
routable[label] = obj
data["commonAnnotations"][label] = resolve_domain(obj)
return routable, data
@cached_property
def json(self):
return json.loads(self.body)
class Meta:
permissions = [("process_alert", "Can process alerts")]
[docs]
class AlertError(models.Model):
alert = models.ForeignKey(Alert, on_delete=models.CASCADE)
created = models.DateTimeField(default=timezone.now)
message = models.TextField()
[docs]
class Audit(models.Model):
body = models.TextField()
created = models.DateTimeField()
data = models.TextField(blank=True)
old = models.TextField(blank=True)
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True)
object_id = models.PositiveIntegerField(default=0)
content_object = GenericForeignKey("content_type", "object_id")
parent_content_type_id = models.PositiveIntegerField(default=0)
parent_object_id = models.PositiveIntegerField(default=0)
user = models.ForeignKey(
settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, default=None
)
class Meta:
indexes = [
models.Index(fields=["parent_content_type_id", "parent_object_id"]),
]
@property
def highlight(self):
if self.body.startswith("Created"):
return "success"
if self.body.startswith("Updated"):
return "warning"
if self.body.startswith("Deleted"):
return "danger"
return ""
@property
def parent_content_type(self) -> ContentType:
return ContentType.objects.get_for_id(self.parent_content_type_id)
@staticmethod
def get_parent(obj):
# The variable's name of the parent object is different depending on the model.
for attr in ["content_object", "project", "service", "farm"]:
value = getattr(obj, attr, None)
if value:
return value
return None
@classmethod
def log(cls, body, instance=None, old=None, **kwargs):
from promgen.middleware import get_current_user
kwargs["body"] = body
kwargs["created"] = timezone.now()
kwargs["user"] = get_current_user()
parent = None
if instance:
kwargs["content_type"] = ContentType.objects.get_for_model(instance)
kwargs["object_id"] = instance.id
data_dict = model_to_dict(instance)
if isinstance(instance, Sender) and data_dict["alias"]:
data_dict["value"] = "********"
kwargs["data"] = json.dumps(data_dict, sort_keys=True)
parent = cls.get_parent(instance)
if old:
kwargs["old"] = json.dumps(model_to_dict(old), sort_keys=True)
# If the parent was changed, we need an additional audit entry for the old parent.
# This is useful for tracking the change of the parent object
# when checking the audit logs of both the old and the new parent.
old_parent = cls.get_parent(old)
if old_parent and old_parent != parent:
kwargs_clone = kwargs.copy()
kwargs_clone["content_type"] = ContentType.objects.get_for_model(old_parent)
kwargs_clone["object_id"] = old_parent.id
cls.objects.create(**kwargs_clone)
# If we have a parent object, we need to set the parent content type and object id
if parent:
kwargs["parent_content_type_id"] = ContentType.objects.get_for_model(parent).id
kwargs["parent_object_id"] = parent.id
return cls.objects.create(**kwargs)
[docs]
class Prometheus(models.Model):
shard = models.ForeignKey("promgen.Shard", on_delete=models.CASCADE)
host = models.CharField(max_length=128)
port = models.IntegerField()
def __str__(self):
return f"{self.host}:{self.port}"
class Meta:
ordering = ["shard", "host"]
unique_together = (("host", "port"),)
verbose_name_plural = "prometheis"
[docs]
class Metric(models.Model):
name = models.CharField(max_length=128, unique=True, validators=[validators.metricname])
type = models.CharField(max_length=32)
description = models.TextField(blank=True)
labels = models.JSONField(default=list)
def to_metric_family(self):
from prometheus_client.core import CounterMetricFamily, HistogramMetricFamily
if self.type == "counter":
family = CounterMetricFamily(
self.name,
self.description,
labels=self.labels,
)
for item in self.samples.all():
family.add_metric(item.labels.split(","), item.data["value"])
return family
elif self.type == "histogram":
family = HistogramMetricFamily(
self.name,
self.description,
labels=self.labels,
)
for item in self.samples.all():
# Convert list of dicts to list of tuples
buckets = [(key, value) for d in item.data["buckets"] for key, value in d.items()]
family.add_metric(
item.labels.split(","),
buckets,
item.data["sum_value"],
)
return family
else:
raise NotImplementedError(f"Metric type {self.type} not implemented")
[docs]
class MetricSample(models.Model):
metric = models.ForeignKey("Metric", on_delete=models.CASCADE, related_name="samples")
labels = models.CharField(
max_length=512, help_text="List of label values", blank=False, null=False
)
data = models.JSONField(default=dict)
class Meta:
unique_together = (("metric", "labels"),)