Merge branch 'main' into feature/google-workspace-api

8naama · web-flow · commit fb3afc7a68bf · 2025-03-31T13:18:36.000+03:00
diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
@@ -22,7 +22,9 @@ jobs:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Build and push
-        uses: docker/build-push-action@v5
+        uses: docker/build-push-action@v6
+        env:
+          DOCKER_BUILD_SUMMARY: false
         with:
           push: true
           sbom: true
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
@@ -36,9 +36,11 @@ jobs:
         # Set required environment variables
         env:
           LOGZIO_API_TOKEN: ${{ secrets.LOGZIO_API_TOKEN }}
+          LOGZIO_API_TOKEN_2: ${{ secrets.LOGZIO_API_TOKEN_2 }}
           DOCKERHUB_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
           DOCKERHUB_USER: ${{ secrets.DOCKER_USERNAME }}
           LOGZIO_SHIPPING_TOKEN: ${{ secrets.LOGZIO_SHIPPING_TOKEN }}
+          LOGZIO_SHIPPING_TOKEN_2: ${{ secrets.LOGZIO_SHIPPING_TOKEN_2 }}
           AZURE_AD_TENANT_ID: ${{ secrets.AZURE_AD_TENANT_ID }}
           AZURE_AD_CLIENT_ID: ${{ secrets.AZURE_AD_CLIENT_ID }}
           AZURE_AD_SECRET_VALUE: ${{ secrets.AZURE_AD_SECRET_VALUE }}
diff --git a/README.md b/README.md
@@ -307,6 +307,9 @@ And your logzio output under `logzio`:
 | url            | The logzio Listener address | Optional          | `https://listener.logz.io:8071` |
 | token          | The logzio shipping token   | Required          | -                               |
 
+> [!NOTE]
+> To configure multiple outputs, please see [multiple outputs example](./src/output/README.md#multiple-outputs)
+
 #### Example
 ```Yaml
 apis:
diff --git a/e2e_tests/api_e2e_test.py b/e2e_tests/api_e2e_test.py
@@ -75,17 +75,18 @@ def run_main_program(self, config_path, secrets_map, test=False):
         thread.start()
         thread.join(timeout=60)
 
-    def search_logs(self, query):
+    def search_logs(self, query, acc=""):
         """
         Search logs in logzio based on the provided query.
 
         Args:
             query (str): The query string to search for in the logs.
+            acc (str): The account API to use, default "" for LOGZIO_API_TOKEN, "2" for LOGZIO_API_TOKEN_2.
 
         Returns:
             list: A list of log entries that match the query.
         """
-        return search_data(query)
+        return search_data(query, acc)
 
     def module_specific_setup(self):
         """
diff --git a/e2e_tests/apis/test_e2e_azure.py b/e2e_tests/apis/test_e2e_azure.py
@@ -1,3 +1,4 @@
+import os
 import time
 
 from e2e_tests.api_e2e_test import ApiE2ETest
@@ -20,16 +21,18 @@ def test_azure_data_in_logz(self):
             "apis.0.azure_ad_tenant_id": "AZURE_AD_TENANT_ID",
             "apis.0.azure_ad_client_id": "AZURE_AD_CLIENT_ID",
             "apis.0.azure_ad_secret_value": "AZURE_AD_SECRET_VALUE",
-            "logzio.token": "LOGZIO_SHIPPING_TOKEN",
+            "logzio.0.token": "LOGZIO_SHIPPING_TOKEN",
+            "logzio.1.token": "LOGZIO_SHIPPING_TOKEN_2",
             "apis.0.additional_fields.type": "TEST_TYPE"
         }
         curr_path = abspath(dirname(__file__))
         config_path = f"{curr_path}/testdata/azure_api_conf.yaml"
         self.run_main_program(config_path=config_path, secrets_map=secrets_map)
         time.sleep(120)
         logs = self.search_logs(f"type:{self.test_type}")
-        self.assertTrue(logs)
-
+        self.assertTrue(logs, "No logs found in logzio account1")
+        logs2 = self.search_logs(f"type:{self.test_type}", "2")
+        self.assertTrue(logs2, "No logs found in logzio account2")
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/e2e_tests/apis/testdata/azure_api_conf.yaml b/e2e_tests/apis/testdata/azure_api_conf.yaml
@@ -12,5 +12,10 @@ apis:
     scrape_interval: 10
 
 logzio:
-  url: https://listener.logz.io:8071
-  token: SHipPIngtoKen
+  - url: https://listener.logz.io:8071
+    token: SHipPIngtoKen
+    inputs: [ azure test ]
+  - url: https://listener.logz.io:8071
+    token: SHipPIngtoKen2
+    inputs:
+      - azure test
diff --git a/e2e_tests/utils/log_utils.py b/e2e_tests/utils/log_utils.py
@@ -3,15 +3,20 @@
 import requests
 
 
-def search_data(query):
+def search_data(query, account=""):
     """
     Send given search query to logzio and returns the result.
     :param query:
+    :param account: the account API to use, "" for LOGZIO_API_TOKEN, "2" for LOGZIO_API_TOKEN_2
     :return:
     """
     url = "https://api.logz.io/v1/search"
+    account_env = {
+        "": "LOGZIO_API_TOKEN",
+        "2": "LOGZIO_API_TOKEN_2"
+    }
     headers = {
-        "X-API-TOKEN": os.environ["LOGZIO_API_TOKEN"],
+        "X-API-TOKEN": os.environ[account_env.get(account)],
         "CONTENT-TYPE": "application/json",
         "ACCEPT": "application/json"
     }
diff --git a/requirements.txt b/requirements.txt
@@ -1,5 +1,5 @@
-pydantic~=2.7.1
-pyyaml~=6.0.1
-requests~=2.32.2
+pydantic~=2.10.6
+pyyaml~=6.0.2
+requests~=2.32.3
 google~=3.0.0
 google-auth~=2.38.0
diff --git a/src/apis/cloudflare/Cloudflare.py b/src/apis/cloudflare/Cloudflare.py
@@ -49,7 +49,8 @@ def __init__(self, **data):
 
         # Update the cloudflare account id in both the url and next url
         self.url = self.url.replace("{account_id}", self.cloudflare_account_id)
-        self.next_url = self.next_url.replace("{account_id}", self.cloudflare_account_id)
+        if self.next_url:
+            self.next_url = self.next_url.replace("{account_id}", self.cloudflare_account_id)
 
         if self.days_back_fetch > 0:
             self._initialize_url_date()
diff --git a/src/apis/general/Api.py b/src/apis/general/Api.py
@@ -38,6 +38,7 @@ class ApiFetcher(BaseModel):
     :param scrape_interval_minutes: the interval between scraping jobs.
     :param url_vars: Not passed to the class, array of params that is generated based on next_url.
     :param body_vars: Not passed to the class, array of params that is generated based on next_body.
+    :param outputs: Not passed to the class, array of outputs to export the returned data to.
     """
     name: str = Field(default="")
     url: str
@@ -52,6 +53,7 @@ class ApiFetcher(BaseModel):
     scrape_interval_minutes: int = Field(default=1, alias="scrape_interval", ge=1)
     url_vars: list = Field(default=[], init=False, init_var=True)
     body_vars: list = Field(default=[], init=False, init_var=True)
+    outputs: list = Field(default=[], init=False, init_var=True)
 
     def __init__(self, **data):
         """
diff --git a/src/apis/oauth/OAuth.py b/src/apis/oauth/OAuth.py
@@ -22,6 +22,7 @@ class OAuthApi(BaseModel):
     :param scrape_interval_minutes: the interval between scraping jobs.
     :param token: The access token, generated by the class after the first request call.
     :param token_expire: The access token expiration time in UNIX, generated by the class after the first request call.
+    :param outputs: Not passed to the class, array of outputs to export the returned data to.
     """
     name: str = Field(default="oauth")
     token_request: ApiFetcher
@@ -30,6 +31,7 @@ class OAuthApi(BaseModel):
     additional_fields: dict = Field(default={})
     token: str = Field(default=None, init=False, init_var=True)
     token_expire: float = Field(default=0, init=False, init_var=True)
+    outputs: list = Field(default=[], init=False, init_var=True)
 
     @model_validator(mode='after')
     def _check_headers(self):
diff --git a/src/config/ConfigReader.py b/src/config/ConfigReader.py
@@ -43,7 +43,7 @@ def __init__(self, conf_file):
         :param conf_file: path to the config file
         """
         self.config = self._read_config(conf_file)
-        self.api_instances, self.logzio_shipper = self.generate_instances()
+        self.api_instances = self.generate_instances()
 
     @staticmethod
     def _read_config(conf_file):
@@ -70,16 +70,15 @@ def generate_instances(self):
         :return: API fetcher (ApiFetcher) instances
         """
         api_instances = []
-        logzio_shipper_instance = None
         shipper_cls = globals().get("LogzioShipper")
         if not self.config:
-            return api_instances, logzio_shipper_instance
+            return api_instances
         apis = self.config.get(INPUT_API_FIELD)
         logzio_conf = self.config.get(OUTPUT_LOGZIO_FIELD)
 
         if not apis:
             logger.error(f"No inputs defined. Please make sure your API input is configured under '{INPUT_API_FIELD}'")
-            return api_instances, logzio_shipper_instance
+            return api_instances
 
         # Generate API fetchers
         for api_conf in apis:
@@ -98,9 +97,34 @@ def generate_instances(self):
                            f"{OUTPUT_LOGZIO_FIELD}")
         else:
             try:
-                logzio_shipper_instance = shipper_cls(**logzio_conf)
-                logger.debug("Created logzio shipper.")
+                # Single output
+                if isinstance(logzio_conf, dict):
+                    logger.debug("Recognized a single output for all APIs.")
+                    logzio_shipper_instance = shipper_cls(**logzio_conf)
+                    logger.debug("Created logzio shipper.")
+                    for api in api_instances:
+                        api.outputs.append(logzio_shipper_instance)
+
+                # Multiple outputs
+                elif isinstance(logzio_conf, list):
+                    logger.debug("Recognized multiple outputs configuration.")
+                    for account in logzio_conf:
+                        logzio_shipper_instance = shipper_cls(**account)
+
+                        if logzio_shipper_instance.inputs:
+                            logger.debug(f"Created logzio shipper for inputs: {logzio_shipper_instance.inputs}")
+                            for api in api_instances:
+                                if api.name in logzio_shipper_instance.inputs:
+                                    api.outputs.append(logzio_shipper_instance)
+                                    logger.debug(f"Assigned logzio shipper for API {api.name}.")
+                        else:
+                            logger.warning(f"Detected a Logzio shipper configuration without any defined inputs. No API data will be exported to it.")
+
+                # Invalid config
+                else:
+                    logger.error(f"Invalid Logzio output config. Please make sure your Logzio config is an object for single output or a list for multiple outputs.")
+
             except (ValidationError, TypeError) as e:
                 logger.error(f"Failed to create Logzio shipper for config {logzio_conf} due to error: {e}")
 
-        return api_instances, logzio_shipper_instance
+        return api_instances
diff --git a/src/main.py b/src/main.py
@@ -56,7 +56,7 @@ def main(conf_path=FETCHER_CONFIG_PATH, test=False):
     conf = ConfigReader(conf_path)
 
     if conf.api_instances:
-        TaskManager(apis=conf.api_instances, logzio_shipper=conf.logzio_shipper).run()
+        TaskManager(apis=conf.api_instances).run()
 
 
 if __name__ == '__main__':
diff --git a/src/manager/TaskManager.py b/src/manager/TaskManager.py
@@ -13,11 +13,9 @@ class TaskManager:
     """
     Class to run scheduled task that collects data from given APIs and sends them with the given logzio_shipper.
     :param apis: List of ApiFetcher instances to fetch data from
-    :param logzio_shipper: LogzioShipper instance to send data to
     """
-    def __init__(self, apis=[], logzio_shipper=None):
+    def __init__(self, apis=[]):
         self.apis = apis
-        self.logzio_shipper = logzio_shipper
         self.threads = []
         self.event = threading.Event()
 
@@ -31,14 +29,14 @@ def _run_api_task(self, api):
         :param api: The API class instance
         """
         logger.info(f"Starting task for api {api.name}.")
-        data_exists = False
 
         try:
-            for log in api.send_request():
-                data_exists = True
-                self.logzio_shipper.add_log_to_send(log, api.additional_fields)
-            if data_exists:
-                self.logzio_shipper.send_to_logzio()
+            logs = api.send_request()
+            if logs:
+                for logzio_shipper in api.outputs:
+                    for log in logs:
+                        logzio_shipper.add_log_to_send(log, api.additional_fields)
+                    logzio_shipper.send_to_logzio()
 
         except requests.exceptions.InvalidURL as e:
             logger.error(f"Failed to send data to Logz.io... Invalid url: {e}")
diff --git a/src/output/LogzioShipper.py b/src/output/LogzioShipper.py
@@ -34,6 +34,7 @@ class LogzioShipper(BaseModel):
     """
     listener: str = Field(default="https://listener.logz.io:8071", alias="url")
     token: str = Field(frozen=True)
+    inputs: list = Field(default=[], frozen=True)
     curr_logs: list = Field(default=[], init=False, init_var=True)
     curr_bulk_size: int = Field(default=0, init=False, init_var=True)
 
diff --git a/src/output/README.md b/src/output/README.md
@@ -0,0 +1,36 @@
+# Logzio output
+It's possible to configure all APIs data to be exported to a single Logz.io account, or to set up multiple outputs and specify which input API to send to which Logz.io account.
+
+## Single output
+```yaml
+logzio:
+  url: https://listener-eu.logz.io:8071  # for us-east-1 region delete url param (default) 
+  token: <<SHIPPING_TOKEN>>
+```
+
+## Multiple outputs
+```yaml
+apis:
+  - name: input 1
+    ...
+  - name: input 2
+    ...
+
+logzio:
+  - url: https://listener-eu.logz.io:8071  # for us-east-1 region delete url param (default)
+   token: <<SHIPPING_TOKEN_ACCOUNT1>>
+   inputs: 
+     - input 1
+     - input 2
+
+  - url: https://listener-eu.logz.io:8071  # for us-east-1 region delete url param (default)
+    token: <<SHIPPING_TOKEN_ACCOUNT2>>
+    inputs: [ "input 1" ]
+```
+
+## Configuration options
+
+| Parameter Name | Description                 | Required/Optional | Default                         |
+|----------------|-----------------------------|-------------------|---------------------------------|
+| url            | The logzio Listener address | Optional          | `https://listener.logz.io:8071` |
+| token          | The logzio shipping token   | Required          | -                               |
diff --git a/tests/UnitTests/test_config_reader.py b/tests/UnitTests/test_config_reader.py
@@ -12,9 +12,9 @@ class TestConfigReader(unittest.TestCase):
     Test reading config from YAML
     """
 
-    def test_invalid_config(self):
+    def test_invalid_input_config(self):
         with self.assertLogs("src.config.ConfigReader", level='INFO') as log:
-            ConfigReader(f"{curr_path}/testConfigs/invalid_conf.yaml")
+            ConfigReader(f"{curr_path}/testConfigs/invalid_input_conf.yaml")
         self.assertIn("ERROR:src.config.ConfigReader:No inputs defined. Please make sure your API input is configured under 'apis'", log.output)
 
     def test_multiple_apis(self):
@@ -23,15 +23,39 @@ def test_multiple_apis(self):
         self.assertEqual(len(conf.api_instances), 2)
         self.assertEqual(conf.api_instances[0].pagination_settings.max_calls, 20)
         self.assertEqual(conf.api_instances[1].scrape_interval_minutes, 5)
+        for a in conf.api_instances:
+            self.assertEqual(len(a.outputs), 1)
 
     def test_missing_logzio_output(self):
         with self.assertLogs("src.config.ConfigReader", level='INFO') as log:
             conf = ConfigReader(f"{curr_path}/testConfigs/missing_logz_output_conf.yaml")
         self.assertIn("WARNING:src.config.ConfigReader:No Logzio shipper output defined. Please make sure your Logzio config is configured under logzio", log.output)
-        self.assertEqual(conf.logzio_shipper, None)
+
+        for a in conf.api_instances:
+            self.assertEqual(a.outputs, [])
 
     def test_open_config_file_fail(self):
         with self.assertLogs("src.config.ConfigReader", level='INFO') as log:
             conf = ConfigReader(f"./not/existing/conf.yaml")
         self.assertEqual(conf.api_instances, [])
-        self.assertEqual(conf.logzio_shipper, None)
+        self.assertIn("ERROR:src.config.ConfigReader:Did not find file ./not/existing/conf.yaml.", log.output)
+
+
+    def test_multiple_outputs(self):
+        with self.assertLogs("src.config.ConfigReader", level='DEBUG') as log:
+            conf = ConfigReader(f"{curr_path}/testConfigs/multiple_outputs_conf.yaml")
+
+        self.assertEqual(len(conf.api_instances), 2)
+        self.assertEqual(len(conf.api_instances[0].outputs), 2)
+        self.assertEqual(len(conf.api_instances[1].outputs), 1)
+
+    def test_multiple_outputs_missing_inputs_conf(self):
+        with self.assertLogs("src.config.ConfigReader", level='INFO') as log:
+            conf = ConfigReader(f"{curr_path}/testConfigs/multiple_outputs_missing_input_conf.yaml")
+        self.assertIn("WARNING:src.config.ConfigReader:Detected a Logzio shipper configuration without any defined inputs. No API data will be exported to it.", log.output)
+
+
+    def test_invalid_output_config(self):
+        with self.assertLogs("src.config.ConfigReader", level='INFO') as log:
+            ConfigReader(f"{curr_path}/testConfigs/invalid_output_conf.yaml")
+        self.assertIn("ERROR:src.config.ConfigReader:Invalid Logzio output config. Please make sure your Logzio config is an object for single output or a list for multiple outputs.", log.output)
diff --git a/tests/testConfigs/invalid_input_conf.yaml b/tests/testConfigs/invalid_input_conf.yaml
diff --git a/tests/testConfigs/invalid_output_conf.yaml b/tests/testConfigs/invalid_output_conf.yaml
@@ -0,0 +1,6 @@
+apis:
+  - name: input1
+    type: general
+    url: https://some/url
+
+logzio: "invalid conf"
diff --git a/tests/testConfigs/multiple_outputs_conf.yaml b/tests/testConfigs/multiple_outputs_conf.yaml
diff --git a/tests/testConfigs/multiple_outputs_missing_input_conf.yaml b/tests/testConfigs/multiple_outputs_missing_input_conf.yaml
