chore: event changes (#2227)

* chore: change max_retries to 10

* chore: only send custom domain deleted event if it is not a partner domain

* chore: newrelic events metric names rename

* chore: emit failed events metric

* chore: migration for contact.flags and custom_domain.pending_deletion

* chore: mark custom_domain as pending_deletion when deleting it

* chore: add event type to metric

app/dashboard/views/domain_detail.py

@@ -267,6 +267,7 @@ def domain_detail(custom_domain_id):
 
             # Schedule delete domain job
             LOG.w("schedule delete domain job for %s", custom_domain)
+            custom_domain.pending_deletion = True
             Job.create(
                 name=JOB_DELETE_DOMAIN,
                 payload={"custom_domain_id": custom_domain.id},

app/events/event_dispatcher.py

@@ -70,9 +70,7 @@ class EventDispatcher:
 
         partner_user = EventDispatcher.__partner_user(user.id)
         if not partner_user:
-            LOG.i(
-                f"Not sending events because there's no partner user for  user {user}"
-            )
+            LOG.i(f"Not sending events because there's no partner user for user {user}")
             return
 
         event = event_pb2.Event(
@@ -84,7 +82,9 @@ class EventDispatcher:
 
         serialized = event.SerializeToString()
         dispatcher.send(serialized)
-        newrelic.agent.record_custom_metric("Custom/events_stored", 1)
+
+        event_type = content.WhichOneof("content")
+        newrelic.agent.record_custom_event("EventStoredToDb", {"type": event_type})
         LOG.i("Sent event to the dispatcher")
 
     @staticmethod

app/models.py

@@ -1863,6 +1863,8 @@ class Contact(Base, ModelMixin):
 
     MAX_NAME_LENGTH = 512
 
+    FLAG_PARTNER_CREATED = 1 << 0
+
     __tablename__ = "contact"
 
     __table_args__ = (
@@ -1921,6 +1923,9 @@ class Contact(Base, ModelMixin):
     # whether contact is created automatically during the forward phase
     automatic_created = sa.Column(sa.Boolean, nullable=True, default=False)
 
+    # contact flags
+    flags = sa.Column(sa.Integer, nullable=False, default=0, server_default="0")
+
     @property
     def email(self):
         return self.website_email
@@ -2427,6 +2432,10 @@ class CustomDomain(Base, ModelMixin):
         server_default=None,
     )
 
+    pending_deletion = sa.Column(
+        sa.Boolean, nullable=False, default=False, server_default="0"
+    )
+
     __table_args__ = (
         Index(
             "ix_unique_domain",  # Index name

event_listener.py

@@ -9,7 +9,7 @@ from events.runner import Runner
 from events.event_source import DeadLetterEventSource, PostgresEventSource
 from events.event_sink import ConsoleEventSink, HttpEventSink
 
-_DEFAULT_MAX_RETRIES = 100
+_DEFAULT_MAX_RETRIES = 10
 
 
 class Mode(Enum):

events/event_sink.py

@@ -27,7 +27,9 @@ class HttpEventSink(EventSink):
             headers={"Content-Type": "application/x-protobuf"},
             verify=not EVENT_WEBHOOK_SKIP_VERIFY_SSL,
         )
-        newrelic.agent.record_custom_event("event_sent", {"http_code": res.status_code})
+        newrelic.agent.record_custom_event(
+            "EventSentToPartner", {"http_code": res.status_code}
+        )
         if res.status_code != 200:
             LOG.warning(
                 f"Failed to send event to webhook: {res.status_code} {res.text}"

job_runner.py

@@ -3,7 +3,7 @@ Run scheduled jobs.
 Not meant for running job at precise time (+- 1h)
 """
 import time
-from typing import List
+from typing import List, Optional
 
 import arrow
 from sqlalchemy.sql.expression import or_, and_
@@ -240,7 +240,7 @@ def process_job(job: Job):
 
     elif job.name == config.JOB_DELETE_DOMAIN:
         custom_domain_id = job.payload.get("custom_domain_id")
-        custom_domain = CustomDomain.get(custom_domain_id)
+        custom_domain: Optional[CustomDomain] = CustomDomain.get(custom_domain_id)
         if not custom_domain:
             return
 
@@ -252,16 +252,17 @@ def process_job(job: Job):
 
         LOG.d("Domain %s deleted", domain_name)
 
-        send_email(
-            user.email,
-            f"Your domain {domain_name} has been deleted",
-            f"""Domain {domain_name} along with its aliases are deleted successfully.
+        if custom_domain.partner_id is None:
+            send_email(
+                user.email,
+                f"Your domain {domain_name} has been deleted",
+                f"""Domain {domain_name} along with its aliases are deleted successfully.
 
-Regards,
-SimpleLogin team.
-""",
-            retries=3,
-        )
+    Regards,
+    SimpleLogin team.
+    """,
+                retries=3,
+            )
     elif job.name == config.JOB_SEND_USER_REPORT:
         export_job = ExportUserDataJob.create_from_job(job)
         if export_job:

migrations/versions/2024_091915_88dd7a0abf54_contact_flags_custom_domain_pending_deletion.py

@@ -0,0 +1,31 @@
+"""contact.flags and custom_domain.pending_deletion
+
+Revision ID: 88dd7a0abf54
+Revises: 2441b7ff5da9
+Create Date: 2024-09-19 15:41:20.910374
+
+"""
+import sqlalchemy_utils
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '88dd7a0abf54'
+down_revision = '2441b7ff5da9'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('contact', sa.Column('flags', sa.Integer(), server_default='0', nullable=False))
+    op.add_column('custom_domain', sa.Column('pending_deletion', sa.Boolean(), server_default='0', nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('custom_domain', 'pending_deletion')
+    op.drop_column('contact', 'flags')
+    # ### end Alembic commands ###

monitoring.py

@@ -125,6 +125,21 @@ def log_events_pending_dead_letter():
     )
 
 
+@newrelic.agent.background_task()
+def log_failed_events():
+    r = Session.execute(
+        """
+        SELECT COUNT(*)
+        FROM sync_event
+        WHERE retries >= 10;
+        """,
+    )
+    failed_events = list(r)[0][0]
+
+    LOG.d("number of failed events %s", failed_events)
+    newrelic.agent.record_custom_metric("Custom/sync_events_failed", failed_events)
+
+
 if __name__ == "__main__":
     exporter = MetricExporter(get_newrelic_license())
     while True:
@@ -132,6 +147,7 @@ if __name__ == "__main__":
         log_nb_db_connection()
         log_pending_to_process_events()
         log_events_pending_dead_letter()
+        log_failed_events()
         Session.close()
 
         exporter.run()