Merge branch 'dev'

Author: peterbull

README.md

@@ -186,7 +186,6 @@ Bodhi-Cast is under the Apache License 2.0. See the [LICENSE](LICENSE) file for
 - [ ] Get shore wind forecast
 - [x] Change data fetching utils to classes
 - [x] Add tests for all endpoints
-- [ ] Add API auth
 - [x] Add user spot creation endpoint
 
 ### Airflow:
@@ -208,3 +207,14 @@ Bodhi-Cast is under the Apache License 2.0. See the [LICENSE](LICENSE) file for
 - [x] Configure DAGs for production
 - [x] Set up env for production
 - [x] Add auto tests for data verification as DAGs
+
+### Week Ending 240218
+
+- [ ] Remove any excessive logging in DAGs
+- [ ] Incoroporate realtime data from available weather stations
+  - [ ] add stations to new table in postgres
+  - [ ] create producer to push to kafka topic
+  - [ ] create consumer to write latest to redis
+  - [ ] add nearby realtime data to spot pages
+  - [ ] add endpoint to find stations within the radius of a given spot
+  - [ ] display nearby station data on frontend

airflow/dags/gefs_wave_etl_from_kafka.py

@@ -8,6 +8,7 @@
 import requests
 import xarray as xr
 from airflow.decorators import task
+from airflow.exceptions import AirflowFailException
 from airflow.operators.empty import EmptyOperator
 from airflow.sensors.external_task import ExternalTaskSensor
 from confluent_kafka import Consumer, KafkaException
@@ -37,8 +38,8 @@
     "email": ["[email protected]"],
     "email_on_failure": False,
     "email_on_retry": False,
-    "retries": 1,
-    "retry_delay": pendulum.duration(minutes=5),
+    "retries": 5,
+    "retry_delay": pendulum.duration(minutes=1),
 }
 
 
@@ -192,7 +193,8 @@ def df_to_db(df, engine, table_name):
                 dtype={"location": Geography(geometry_type="POINT", srid=4326)},
             )
             entry_id = df["valid_time"].unique()
-            entry_id = entry_id[0].strftime("%Y-%m-%d %H:%M:%S")
+            entry_id_datetime = pd.to_datetime(entry_id[0])
+            entry_id_str = entry_id_datetime.strftime("%Y-%m-%d %H:%M:%S")
             print(f"Successfully wrote grib2 file for {entry_id}")
         except SQLAlchemyError as e:
             print(f"An error occurred: {e}")
@@ -207,30 +209,65 @@ def df_to_db(df, engine, table_name):
     is_paused_upon_creation=False,
 ) as dag:
 
-    # ExternalTaskSensor to wait for gefs_wave_urls_to_kafka DAG to complete
-    wait_for_gefs_wave_urls_to_kafka = ExternalTaskSensor(
-        task_id="wait_for_gefs_wave_urls_to_kafka",
-        external_dag_id="gefs_wave_urls_to_kafka",
-        external_task_id=None,  # `None` will wait for the entire task to complete
-        timeout=7200,  # Timeout before failing task
-        poke_interval=60,  # Seconds to wait between checks
-        mode="reschedule",  # Reschedule for long waits to free up worker slots
-    )
-
-    @task
-    def consume_and_process_from_kafka():
-        consume_from_kafka(
-            topic=topic,
-            engine=engine,
-            table_name=table_name,
-            bs=8,
-            sasl_username=sasl_username,
-            sasl_password=sasl_password,
-        )
+    def taskflow():
+        conf = {
+            "bootstrap.servers": "kafka:9092",
+            "group.id": "airflow-consumers",
+            "enable.auto.commit": False,
+            "auto.offset.reset": "earliest",  # consume from the start of topic
+            "security.protocol": "SASL_PLAINTEXT",
+            "sasl.mechanisms": "PLAIN",
+            "sasl.username": sasl_username,
+            "sasl.password": sasl_password,
+            "max.poll.interval.ms": 900000,
+        }
+
+        # # ExternalTaskSensor to wait for gefs_wave_urls_to_kafka DAG to complete
+        # wait_for_gefs_wave_urls_to_kafka = ExternalTaskSensor(
+        #     task_id="wait_for_gefs_wave_urls_to_kafka",
+        #     external_dag_id="gefs_wave_urls_to_kafka",
+        #     external_task_id=None,  # `None` will wait for the entire task to complete
+        #     timeout=7200,  # Timeout before failing task
+        #     poke_interval=60,  # Seconds to wait between checks
+        #     mode="reschedule",  # Reschedule for long waits to free up worker slots
+        # )
+
+        @task
+        def check_for_messages():
+            c = Consumer(conf)
+            c.subscribe([topic])
+            logging.info(f"{conf}")
+            # Poll for messages
+            msg = c.poll(30.0)
+            c.close()
+
+            if msg is None:
+                logging.info("No new messages found. Task will be retried.")
+                raise AirflowFailException(
+                    "No new messages found. Task will be explicitly failed to trigger retry."
+                )
+            else:
+                logging.info("New messages found. Proceeding to consume and process.")
+                return True
+
+        @task
+        def consume_and_process_from_kafka():
+            consume_from_kafka(
+                topic=topic,
+                engine=engine,
+                table_name=table_name,
+                bs=8,
+                sasl_username=sasl_username,
+                sasl_password=sasl_password,
+            )
+
+        check_result = check_for_messages()
+        consume_task = consume_and_process_from_kafka()
+        check_result >> consume_task
 
-    data = consume_and_process_from_kafka()
+        # wait_for_gefs_wave_urls_to_kafka >> data
 
-    wait_for_gefs_wave_urls_to_kafka >> data
+    dag = taskflow()
 
 if __name__ == "__main__":
     dag.test()

airflow/dags/gefs_wave_urls_to_kafka.py

@@ -8,7 +8,6 @@
 from airflow.decorators import task
 from bs4 import BeautifulSoup
 from confluent_kafka import Consumer, KafkaException, Producer
-from confluent_kafka.admin import AdminClient, ConfigResource
 
 from airflow import DAG
 
@@ -26,7 +25,7 @@
     "email": ["[email protected]"],
     "email_on_failure": False,
     "email_on_retry": False,
-    "retries": 1,
+    "retries": 5,
     "retry_delay": pendulum.duration(minutes=5),
 }
 

airflow/dags/monitor_gefs_urls_kafka.py

@@ -0,0 +1,77 @@
+import logging
+import os
+
+import pendulum
+from airflow.decorators import task
+from confluent_kafka import Consumer, KafkaException
+
+from airflow import DAG
+
+sasl_username = os.environ.get("KAFKA_DEFAULT_USERS")
+sasl_password = os.environ.get("KAFKA_DEFAULT_PASSWORDS")
+
+
+start_date = pendulum.datetime(2024, 1, 1)
+
+default_args = {
+    "owner": "airflow",
+    "depends_on_past": False,
+    "start_date": start_date,
+    "email": ["[email protected]"],
+    "email_on_failure": False,
+    "email_on_retry": False,
+    "retries": 1,
+    "retry_delay": pendulum.duration(minutes=5),
+}
+
+
+with DAG(
+    "monitor_kafka_topics",
+    default_args=default_args,
+    description="Monitor the content of Kafka topics",
+    schedule=None,
+    catchup=False,
+) as dag:
+
+    def taskflow():
+
+        @task
+        def monitor_kafka_topic(topic, sasl_username=sasl_username, sasl_password=sasl_password):
+            conf = {
+                "bootstrap.servers": "kafka:9092",
+                "group.id": "airflow-consumers",
+                "enable.auto.commit": False,
+                "auto.offset.reset": "earliest",  # consume from the start of topic
+                "security.protocol": "SASL_PLAINTEXT",
+                "sasl.mechanisms": "PLAIN",
+                "sasl.username": sasl_username,
+                "sasl.password": sasl_password,
+            }
+
+            c = Consumer(conf)
+
+            c.subscribe([topic])
+
+            while True:
+                msg = c.poll(9.0)
+                if msg is None:
+                    logging.info(f"No more messages in topic {topic}")
+                    break
+                if msg.error():
+                    logging.error(f"Error consuming from topic {topic}: {msg.error()}")
+                    raise KafkaException(msg.error())
+                logging.info("Received message: {}".format(msg.value().decode("utf-8")))
+            # Skipping committing the offsets so that this can be run as many times as needed
+            c.close()
+
+        monitor_wave_urls = monitor_kafka_topic(
+            topic="gefs_wave_urls", sasl_username=sasl_username, sasl_password=sasl_password
+        )
+
+        monitor_station_data = monitor_kafka_topic(
+            topic="noaa_station_latest_data",
+            sasl_username=sasl_username,
+            sasl_password=sasl_password,
+        )
+
+    dag = taskflow()