From bc75182d0f0901c03f4a71957412b102bf3da61e Mon Sep 17 00:00:00 2001
From: Ivanildo Barauna de Souza Junior <ivanildo.jnr@outlook.com>
Date: Sun, 11 Aug 2024 05:44:36 +0000
Subject: [PATCH 1/3] feat: update to Python 10

---
 Dockerfile           |  2 +-
 pipeline/launcher.py |  6 +++---
 pipeline_runner.sh   | 15 +--------------
 3 files changed, 5 insertions(+), 18 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index b3ddc29..831934a 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM apache/beam_python3.9_sdk:2.57.0
+FROM apache/beam_python3.10_sdk:2.57.0
 
 WORKDIR /app
 
diff --git a/pipeline/launcher.py b/pipeline/launcher.py
index b23b457..2691aef 100644
--- a/pipeline/launcher.py
+++ b/pipeline/launcher.py
@@ -10,8 +10,8 @@
 metrics_client = MetricsClient()
 
 
-def increment_metric(element):
-    metrics_client.incr(metric_name=f"{APPLICATION_NAME}.extracted", value=1)
+def increment_metric(element, metric_name: str):
+    metrics_client.incr(metric_name=f"{APPLICATION_NAME}.{metric_name}", value=1)
     return element
 
 
@@ -20,7 +20,7 @@ def run():
         (
             p
             | "ReadTopic" >> beam.io.ReadFromPubSub(subscription=PUBSUB_SUBSCRIPTION)
-            | "ExtractMetric" >> beam.Map(lambda x: increment_metric(x))
+            | "ExtractMetric" >> beam.Map(lambda x: increment_metric(x, "extracted"))
             | "Decode" >> beam.Map(lambda x: x.decode("utf-8"))
             | "CreateKey" >> beam.Map(set_key, "created_at", 8)
             | "Window" >> beam.WindowInto(beam.window.FixedWindows(5))
diff --git a/pipeline_runner.sh b/pipeline_runner.sh
index c04b209..be051da 100644
--- a/pipeline_runner.sh
+++ b/pipeline_runner.sh
@@ -1,15 +1,3 @@
-## Using setup.py
-
-#python main.py --project ivanildobarauna \
-#               --region us-east4 \
-#               --runner Dataflowrunner \
-#--setup_file ./setup.py \
-#--staging_location gs://gcp-streaming-pipeline-dataflow/staging \
-#--temp_location gs://gcp-streaming-pipeline-dataflow/temp \
-#--save_main_session
-
-## Using container
-
 python main.py --project ivanildobarauna \
                --region us-east4 \
                --runner DataflowRunner \
@@ -20,5 +8,4 @@ python main.py --project ivanildobarauna \
                --worker_machine_type=n1-standard-1 \
                --num_workers=1 \
                --max_num_workers=1 \
-               --autoscaling_algorithm=NONE
-
+               --autoscaling_algorithm=NONE
\ No newline at end of file

From 067b9d4b49019cf1b98071db303f51997e5bb519 Mon Sep 17 00:00:00 2001
From: Ivanildo Barauna de Souza Junior <ivanildo.jnr@outlook.com>
Date: Tue, 13 Aug 2024 10:39:53 -0300
Subject: [PATCH 2/3] feature: increment writed_metric

---
 .../{BuildImage.xml => Build.xml}             |  2 +-
 .idea/runConfigurations/PipelineDeploy.xml    |  8 ++++
 .../runConfigurations/PipelineDeployment.xml  | 17 ++++++++
 .../{PipelineRunner.xml => PipelineTest.xml}  |  4 +-
 .../{PushImage.xml => Push.xml}               |  2 +-
 Dockerfile                                    |  2 +
 main.py                                       |  3 ++
 pipeline/launcher.py                          | 40 +++++++++++++------
 pipeline/utils/send_metrics.py                |  9 ++++-
 pipeline_runner.sh => pipeline_deployment.sh  |  0
 pipeline_test.sh                              | 11 +++++
 11 files changed, 80 insertions(+), 18 deletions(-)
 rename .idea/runConfigurations/{BuildImage.xml => Build.xml} (77%)
 create mode 100644 .idea/runConfigurations/PipelineDeploy.xml
 create mode 100644 .idea/runConfigurations/PipelineDeployment.xml
 rename .idea/runConfigurations/{PipelineRunner.xml => PipelineTest.xml} (81%)
 rename .idea/runConfigurations/{PushImage.xml => Push.xml} (91%)
 rename pipeline_runner.sh => pipeline_deployment.sh (100%)
 create mode 100644 pipeline_test.sh

diff --git a/.idea/runConfigurations/BuildImage.xml b/.idea/runConfigurations/Build.xml
similarity index 77%
rename from .idea/runConfigurations/BuildImage.xml
rename to .idea/runConfigurations/Build.xml
index ae1be5b..af720d9 100644
--- a/.idea/runConfigurations/BuildImage.xml
+++ b/.idea/runConfigurations/Build.xml
@@ -1,5 +1,5 @@
 <component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="BuildImage" type="docker-deploy" factoryName="dockerfile" server-name="Docker">
+  <configuration default="false" name="Build" type="docker-deploy" factoryName="dockerfile" server-name="Docker">
     <deployment type="dockerfile">
       <settings>
         <option name="imageTag" value="us-central1-docker.pkg.dev/ivanildobarauna/dataflow-ar/custom-beam/img:latest" />
diff --git a/.idea/runConfigurations/PipelineDeploy.xml b/.idea/runConfigurations/PipelineDeploy.xml
new file mode 100644
index 0000000..f012be3
--- /dev/null
+++ b/.idea/runConfigurations/PipelineDeploy.xml
@@ -0,0 +1,8 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="PipelineDeploy" type="CompoundRunConfigurationType">
+    <toRun name="Build" type="docker-deploy" />
+    <toRun name="PipelineDeployment" type="ShConfigurationType" />
+    <toRun name="Push" type="ShConfigurationType" />
+    <method v="2" />
+  </configuration>
+</component>
\ No newline at end of file
diff --git a/.idea/runConfigurations/PipelineDeployment.xml b/.idea/runConfigurations/PipelineDeployment.xml
new file mode 100644
index 0000000..a79ad6f
--- /dev/null
+++ b/.idea/runConfigurations/PipelineDeployment.xml
@@ -0,0 +1,17 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="PipelineDeployment" type="ShConfigurationType">
+    <option name="SCRIPT_TEXT" value="" />
+    <option name="INDEPENDENT_SCRIPT_PATH" value="true" />
+    <option name="SCRIPT_PATH" value="$PROJECT_DIR$/pipeline_deployment.sh" />
+    <option name="SCRIPT_OPTIONS" value="" />
+    <option name="INDEPENDENT_SCRIPT_WORKING_DIRECTORY" value="true" />
+    <option name="SCRIPT_WORKING_DIRECTORY" value="$PROJECT_DIR$" />
+    <option name="INDEPENDENT_INTERPRETER_PATH" value="true" />
+    <option name="INTERPRETER_PATH" value="/bin/zsh" />
+    <option name="INTERPRETER_OPTIONS" value="" />
+    <option name="EXECUTE_IN_TERMINAL" value="true" />
+    <option name="EXECUTE_SCRIPT_FILE" value="true" />
+    <envs />
+    <method v="2" />
+  </configuration>
+</component>
\ No newline at end of file
diff --git a/.idea/runConfigurations/PipelineRunner.xml b/.idea/runConfigurations/PipelineTest.xml
similarity index 81%
rename from .idea/runConfigurations/PipelineRunner.xml
rename to .idea/runConfigurations/PipelineTest.xml
index f0032c1..42459bf 100644
--- a/.idea/runConfigurations/PipelineRunner.xml
+++ b/.idea/runConfigurations/PipelineTest.xml
@@ -1,8 +1,8 @@
 <component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="PipelineRunner" type="ShConfigurationType">
+  <configuration default="false" name="PipelineTest" type="ShConfigurationType">
     <option name="SCRIPT_TEXT" value="" />
     <option name="INDEPENDENT_SCRIPT_PATH" value="true" />
-    <option name="SCRIPT_PATH" value="$PROJECT_DIR$/pipeline_runner.sh" />
+    <option name="SCRIPT_PATH" value="$PROJECT_DIR$/pipeline_test.sh" />
     <option name="SCRIPT_OPTIONS" value="" />
     <option name="INDEPENDENT_SCRIPT_WORKING_DIRECTORY" value="true" />
     <option name="SCRIPT_WORKING_DIRECTORY" value="$PROJECT_DIR$" />
diff --git a/.idea/runConfigurations/PushImage.xml b/.idea/runConfigurations/Push.xml
similarity index 91%
rename from .idea/runConfigurations/PushImage.xml
rename to .idea/runConfigurations/Push.xml
index 0684b74..aadc8ad 100644
--- a/.idea/runConfigurations/PushImage.xml
+++ b/.idea/runConfigurations/Push.xml
@@ -1,5 +1,5 @@
 <component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="PushImage" type="ShConfigurationType">
+  <configuration default="false" name="Push" type="ShConfigurationType">
     <option name="SCRIPT_TEXT" value="docker push us-central1-docker.pkg.dev/ivanildobarauna/dataflow-ar/custom-beam/img:latest" />
     <option name="INDEPENDENT_SCRIPT_PATH" value="true" />
     <option name="SCRIPT_PATH" value="" />
diff --git a/Dockerfile b/Dockerfile
index 831934a..cd5892d 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -2,6 +2,8 @@ FROM apache/beam_python3.10_sdk:2.57.0
 
 WORKDIR /app
 
+ENV env=production
+
 # Update pip and install poetry
 RUN pip install --upgrade pip \
     && pip install poetry
diff --git a/main.py b/main.py
index 87884c8..3e43d49 100644
--- a/main.py
+++ b/main.py
@@ -1,7 +1,10 @@
+import os
 import logging
 
 from pipeline.launcher import run
 
 if __name__ == "__main__":
+    if not os.getenv("env"):
+        os.environ["env"] = "test"
     logging.getLogger().setLevel(logging.INFO)
     run()
diff --git a/pipeline/launcher.py b/pipeline/launcher.py
index 2691aef..17a102c 100644
--- a/pipeline/launcher.py
+++ b/pipeline/launcher.py
@@ -1,3 +1,4 @@
+import os
 import apache_beam as beam
 from pipeline.config.beam_config import get_options
 from pipeline.config.gcp.constants import PUBSUB_SUBSCRIPTION
@@ -10,17 +11,26 @@
 metrics_client = MetricsClient()
 
 
-def increment_metric(element, metric_name: str):
-    metrics_client.incr(metric_name=f"{APPLICATION_NAME}.{metric_name}", value=1)
-    return element
+class IncrementMetricFn(beam.DoFn):
+    def process(self, element, action: str):
+        metrics_client.incr(metric_name=APPLICATION_NAME, action=action, value=1)
+        # yield element
+
+
+def increment_metric(element: str, action: str):
+    metrics_client.incr(metric_name=APPLICATION_NAME, action=action, value=1)
+
+    if action == "extracted":
+        return element
 
 
 def run():
     with beam.Pipeline(options=get_options()) as p:
-        (
+        data = (
             p
             | "ReadTopic" >> beam.io.ReadFromPubSub(subscription=PUBSUB_SUBSCRIPTION)
-            | "ExtractMetric" >> beam.Map(lambda x: increment_metric(x, "extracted"))
+            | "ExtractMetric"
+            >> beam.Map(lambda x: increment_metric(element=x, action="extracted"))
             | "Decode" >> beam.Map(lambda x: x.decode("utf-8"))
             | "CreateKey" >> beam.Map(set_key, "created_at", 8)
             | "Window" >> beam.WindowInto(beam.window.FixedWindows(5))
@@ -28,11 +38,17 @@ def run():
             | "ExtractMessage" >> beam.Map(lambda x: x[1])
             | "ParseToJson" >> beam.Map(set_datetime, "ARRIVAL_DATE", "data")
             | "Flatten" >> beam.FlatMap(lambda x: x)
-            | "WriteToBigQuery"
-            >> beam.io.WriteToBigQuery(
-                table=STG.api_data.get_table_name(),
-                schema=STG.api_data.schema(),
-                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
-                method="STREAMING_INSERTS",
-            )
+        )
+
+        # Escreve no BigQuery
+        data | "WriteToBigQuery" >> beam.io.WriteToBigQuery(
+            table=STG.api_data.get_table_name(),
+            schema=STG.api_data.schema(),
+            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
+            method="STREAMING_INSERTS",
+        )
+
+        # Incrementa a métrica depois da escrita
+        data | "IncrementFinalMetric" >> beam.ParDo(
+            IncrementMetricFn(), "bigquery_writed"
         )
diff --git a/pipeline/utils/send_metrics.py b/pipeline/utils/send_metrics.py
index 4d20318..ed91ad0 100644
--- a/pipeline/utils/send_metrics.py
+++ b/pipeline/utils/send_metrics.py
@@ -2,6 +2,7 @@
 from pipeline.utils.instance_validator import Singleton
 from pipeline.config.gcp.constants import PROJECT_ID, DD_AGENT_SERVER_SECRET_ID
 from google.cloud import secretmanager
+import os
 
 
 class DataDogClient:
@@ -27,5 +28,9 @@ def __init__(self) -> None:
             DataDogClient()
             self.initialized = True
 
-    def incr(self, metric_name: str, value: int):
-        statsd.increment(metric=metric_name, value=value)
+    def incr(self, metric_name: str, action: str, value: int):
+        statsd.increment(
+            metric=metric_name,
+            value=value,
+            tags=["action:" + action, "env:" + os.getenv("env")],
+        )
diff --git a/pipeline_runner.sh b/pipeline_deployment.sh
similarity index 100%
rename from pipeline_runner.sh
rename to pipeline_deployment.sh
diff --git a/pipeline_test.sh b/pipeline_test.sh
new file mode 100644
index 0000000..44bdde8
--- /dev/null
+++ b/pipeline_test.sh
@@ -0,0 +1,11 @@
+python main.py --project ivanildobarauna \
+               --region us-east4 \
+               --runner DirectRunner \
+               --staging_location gs://gcp-streaming-pipeline-dataflow/staging \
+               --temp_location gs://gcp-streaming-pipeline-dataflow/temp \
+               --sdk_container_image=us-central1-docker.pkg.dev/ivanildobarauna/dataflow-ar/custom-beam/img:latest \
+               --sdk_location=container \
+               --worker_machine_type=n1-standard-1 \
+               --num_workers=1 \
+               --max_num_workers=1 \
+               --autoscaling_algorithm=NONE
\ No newline at end of file

From 96a78371482533912b7b898592f96c90626a7b2c Mon Sep 17 00:00:00 2001
From: Ivanildo Barauna de Souza Junior <ivanildo.jnr@outlook.com>
Date: Tue, 13 Aug 2024 11:12:58 -0300
Subject: [PATCH 3/3] feature: update to py10

---
 .idea/data-consumer-pipeline.iml | 2 +-
 .idea/misc.xml                   | 2 +-
 pipeline_deployment.sh           | 2 +-
 pipeline_test.sh                 | 2 +-
 4 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/.idea/data-consumer-pipeline.iml b/.idea/data-consumer-pipeline.iml
index 2c80e12..cd246b9 100644
--- a/.idea/data-consumer-pipeline.iml
+++ b/.idea/data-consumer-pipeline.iml
@@ -4,7 +4,7 @@
     <content url="file://$MODULE_DIR$">
       <excludeFolder url="file://$MODULE_DIR$/.venv" />
     </content>
-    <orderEntry type="inheritedJdk" />
+    <orderEntry type="jdk" jdkName="Poetry (data-consumer-pipeline)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
 </module>
\ No newline at end of file
diff --git a/.idea/misc.xml b/.idea/misc.xml
index 536cb0e..4739fd5 100644
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -3,5 +3,5 @@
   <component name="Black">
     <option name="sdkName" value="Python 3.9 (data-consumer-pipeline)" />
   </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9 (data-consumer-pipeline)" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Poetry (data-consumer-pipeline)" project-jdk-type="Python SDK" />
 </project>
\ No newline at end of file
diff --git a/pipeline_deployment.sh b/pipeline_deployment.sh
index be051da..2f8980b 100644
--- a/pipeline_deployment.sh
+++ b/pipeline_deployment.sh
@@ -5,7 +5,7 @@ python main.py --project ivanildobarauna \
                --temp_location gs://gcp-streaming-pipeline-dataflow/temp \
                --sdk_container_image=us-central1-docker.pkg.dev/ivanildobarauna/dataflow-ar/custom-beam/img:latest \
                --sdk_location=container \
-               --worker_machine_type=n1-standard-1 \
+               --worker_machine_type=e2-custom-1-3072 \
                --num_workers=1 \
                --max_num_workers=1 \
                --autoscaling_algorithm=NONE
\ No newline at end of file
diff --git a/pipeline_test.sh b/pipeline_test.sh
index 44bdde8..dedcd61 100644
--- a/pipeline_test.sh
+++ b/pipeline_test.sh
@@ -5,7 +5,7 @@ python main.py --project ivanildobarauna \
                --temp_location gs://gcp-streaming-pipeline-dataflow/temp \
                --sdk_container_image=us-central1-docker.pkg.dev/ivanildobarauna/dataflow-ar/custom-beam/img:latest \
                --sdk_location=container \
-               --worker_machine_type=n1-standard-1 \
+               --worker_machine_type=e2-custom-1-3072 \
                --num_workers=1 \
                --max_num_workers=1 \
                --autoscaling_algorithm=NONE
\ No newline at end of file