Add HTTP notifications

Author: woodlee

plan_monitor/config.py

@@ -44,12 +44,16 @@
 EVICTION_THROTTLE_TIME_WINDOW_SECONDS = int(os.environ.get('EVICTION_THROTTLE_TIME_WINDOW_SECONDS', 300))
 EVICTION_THROTTLE_MAX_PLANS_FOR_TIME_WINDOW = int(os.environ.get('EVICTION_THROTTLE_MAX_PLANS_FOR_TIME_WINDOW', 5))
 
-# Settings that you provide if you are using the notify_slack module to send eviction notices:
+# Settings that you provide if you are using the notify module to send eviction notices TO SLACK:
 SLACK_API_TOKEN = os.environ.get('SLACK_API_TOKEN', '')
 SLACK_NOTIFY_CHANNEL = os.environ.get('SLACK_NOTIFY_CHANNEL', '')
 SLACK_MESSAGE_TEMPLATE = os.environ.get('SLACK_MESSAGE_TEMPLATE', '')
 SLACK_POST_AS_BLOCKS = os.environ.get('SLACK_POST_AS_BLOCKS', 'true').lower() in ('1', 't', 'true', 'yes')
 
+# Settings that you provide if you are using the notify module to send eviction notices via HTTP POSTs:
+HTTP_NOTIFY_HEADERS = os.environ.get('HTTP_NOTIFY_HEADERS', '')
+HTTP_NOTIFY_TEMPLATE = os.environ.get('HTTP_NOTIFY_TEMPLATE', '')
+HTTP_NOTIFY_URL = os.environ.get('HTTP_NOTIFY_URL', '')
 
 # Bad plan detection tuning
 # ------------------------------------------------------------------------------------------------------------

plan_monitor/notify_slack.py plan_monitor/notify.pyplan_monitor/notify_slack.py renamed to plan_monitor/notify.py

@@ -7,17 +7,19 @@
 
 import confluent_kafka
 import confluent_kafka.schema_registry.avro
+import requests
+
 from jinja2 import Template
 from slack import WebClient
 from slack.errors import SlackApiError
 
 from . import config, message_schemas, common
 
 
-logger = logging.getLogger('plan_monitor.notify_slack')
+logger = logging.getLogger('plan_monitor.notify')
 
 
-def notify(slack_client: WebClient, msg: confluent_kafka.Message) -> None:
+def notify_slack(slack_client: WebClient, msg: confluent_kafka.Message) -> None:
     prior_plans_count, prior_times_sum, prior_reads_sum, prior_execs_sum, prior_last_execution = 0, 0, 0, 0, 0
     msg_val = dict(msg.value())
     for pp in msg_val['prior_plans']:
@@ -54,15 +56,30 @@ def notify(slack_client: WebClient, msg: confluent_kafka.Message) -> None:
             logger.warning(f"Error sending message to Slack: {e.response.get('error', '<none>')}")
 
 
-def notify_slack() -> None:
+def notify_http(msg: confluent_kafka.Message) -> None:
+    if config.HTTP_NOTIFY_TEMPLATE:
+        template = Template(config.HTTP_NOTIFY_TEMPLATE)
+        body = template.render(msg=msg)
+    else:
+        body = json.dumps(msg)
+
+    headers = json.loads(config.HTTP_NOTIFY_HEADERS) if config.HTTP_NOTIFY_HEADERS else {}
+    resp = requests.post(config.HTTP_NOTIFY_URL, data=body, headers=headers, timeout=5.0)
+    resp.raise_for_status()
+
+    logger.debug('Posted eviction notification to %s with code %s and response: %s', config.HTTP_NOTIFY_URL,
+                 resp.status_code, resp.text)
+
+
+def notify() -> None:
     schema_registry = confluent_kafka.schema_registry.SchemaRegistryClient({'url': config.SCHEMA_REGISTRY_URL})
     key_deserializer = confluent_kafka.schema_registry.avro.AvroDeserializer(
         message_schemas.EVICTED_PLANS_MESSAGE_KEY_AVRO_SCHEMA, schema_registry)
     value_deserializer = confluent_kafka.schema_registry.avro.AvroDeserializer(
         message_schemas.EVICTED_PLANS_MESSAGE_VALUE_AVRO_SCHEMA, schema_registry)
 
     consumer_config = {'bootstrap.servers': config.KAFKA_BOOTSTRAP_SERVERS,
-                       'group.id': f'sqlserver_plan_regression_monitor_notify_slack_{socket.getfqdn()}',
+                       'group.id': f'sqlserver_plan_regression_monitor_notify_{socket.getfqdn()}',
                        'key.deserializer': key_deserializer,
                        'value.deserializer': value_deserializer,
                        'enable.auto.commit': True,
@@ -72,10 +89,12 @@ def notify_slack() -> None:
 
     kafka_consumer = confluent_kafka.DeserializingConsumer(consumer_config)
     kafka_consumer.subscribe([config.EVICTED_PLANS_TOPIC])
-    slack_client = WebClient(token=config.SLACK_API_TOKEN)
     last_log_heartbeat = datetime.utcnow()
     log_heartbeat_interval_seconds = 60
 
+    if config.SLACK_NOTIFY_CHANNEL:
+        slack_client = WebClient(token=config.SLACK_API_TOKEN)
+
     try:
         while True:
             if (datetime.utcnow() - last_log_heartbeat).total_seconds() > log_heartbeat_interval_seconds:
@@ -88,10 +107,17 @@ def notify_slack() -> None:
             if msg is None:
                 continue
 
-            logger.info(f'Notifying Slack channel {config.SLACK_NOTIFY_CHANNEL} for message '
-                        f'@ {common.msg_coordinates(msg)}')
+            if config.SLACK_NOTIFY_CHANNEL:
+                logger.info(f'Notifying Slack channel {config.SLACK_NOTIFY_CHANNEL} for message '
+                            f'@ {common.msg_coordinates(msg)}')
+                notify_slack(slack_client, msg)
+
+            if config.HTTP_NOTIFY_URL:
+                logger.info(f'Notifying via HTTP POST for message '
+                            f'@ {common.msg_coordinates(msg)}')
+                notify_http(msg)
+
             last_log_heartbeat = datetime.utcnow()
-            notify(slack_client, msg)
             kafka_consumer.store_offsets(msg)
     except KeyboardInterrupt:
         logger.info('Received interrupt request; shutting down...')
@@ -102,4 +128,4 @@ def notify_slack() -> None:
 
 if __name__ == "__main__":
     logger.info('Starting...')
-    notify_slack()
+    notify()

requirements.txt

@@ -2,5 +2,6 @@ confluent-kafka[avro]==1.5.0
 pyodbc==4.0.30
 python-dateutil==2.8.1
 pytz==2020.1
+requests==2.26.0
 slackclient==2.8.0
 Jinja2==2.11.2