feat(traefik): add configuration for Traefik V3 API CRD group

README.md

@@ -30,6 +30,7 @@ Before running the controller, make sure to configure the following environment
 - `WATCH_INGRESSROUTES`: Set to `True` to enable monitoring of Traefik IngressRoutes.
 - `WATCH_INGRESS`: Set to `True` to enable monitoring of Kubernetes Ingress resources.
 - `WATCH_INTERVAL`: Interval in seconds between each check for changes in Ingress or IngressRoutes (default is `10` seconds).
+- `USE_TRAEFIK_V3_CRD_GROUP`: Whether to use Traefik V3 API CRD group (`traefik.io`); default to `False`.
 
 ### Annotations for Uptime Kuma Autodiscovery
 
@@ -158,4 +159,4 @@ poetry run pytest
 
 ## License
 
-This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.
+This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.

kuma_ingress_watcher/controller.py

@@ -8,6 +8,8 @@
 
 
 def str_to_bool(value):
+    if isinstance(value, bool):
+        return value
     return str(value).lower() in ['true', '1', 't', 'y', 'yes']
 
 
@@ -16,8 +18,9 @@ def str_to_bool(value):
 UPTIME_KUMA_USER = os.getenv('UPTIME_KUMA_USER')
 UPTIME_KUMA_PASSWORD = os.getenv('UPTIME_KUMA_PASSWORD')
 WATCH_INTERVAL = int(os.getenv('WATCH_INTERVAL', '10') or 10)
-WATCH_INGRESSROUTES = str_to_bool(os.getenv('WATCH_INGRESSROUTES', 'True'))
-WATCH_INGRESS = str_to_bool(os.getenv('WATCH_INGRESS', 'False'))
+WATCH_INGRESSROUTES = str_to_bool(os.getenv('WATCH_INGRESSROUTES', True))
+WATCH_INGRESS = str_to_bool(os.getenv('WATCH_INGRESS', False))
+USE_TRAEFIK_V3_CRD_GROUP = str_to_bool(os.getenv('USE_TRAEFIK_V3_CRD_GROUP', False))
 LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper()
 
 LOG_LEVELS = {
@@ -186,9 +189,14 @@ def init_kubernetes_client():
 
 
 def get_ingressroutes(custom_api_instance):
+    if USE_TRAEFIK_V3_CRD_GROUP:
+        group = "traefik.io"
+    else:
+        group = "traefik.containo.us"
+
     try:
         return custom_api_instance.list_cluster_custom_object(
-            group="traefik.containo.us",
+            group=group,
             version="v1alpha1",
             plural="ingressroutes"
         )

tests/test_get_ingressroutes.py

@@ -19,6 +19,11 @@ def test_get_ingressroutes_success(self, mock_api_instance, mock_logger):
         # Call the function
         result = get_ingressroutes(mock_api_instance)
 
+        # Assert we call list_cluster_custom_object with expected group
+        mock_api_instance.list_cluster_custom_object.assert_called_with(group='traefik.containo.us',
+                                                                        version='v1alpha1',
+                                                                        plural='ingressroutes')
+
         # Expected result after calling 'to_dict' on the ingressroute object
         expected_result = {'items': [{'metadata': {'name': 'test-ingressroute'}}]}
 
@@ -50,6 +55,24 @@ def test_get_ingressroutes_empty(self, mock_api_instance, mock_logger):
         self.assertEqual(result, {'items': []})
         mock_logger.error.assert_not_called()
 
+    @patch('kuma_ingress_watcher.controller.custom_api_instance')
+    @patch('kuma_ingress_watcher.controller.USE_TRAEFIK_V3_CRD_GROUP', True)
+    def test_get_ingressroutes_traefik_v3_crd(self, mock_api_instance):
+        mock_ingressroute = MagicMock()
+        mock_ingressroute.to_dict.return_value = {'metadata': {'name': 'test-ingressroute'}}
+
+        # Mock the API response to include the mocked ingressroute
+        mock_api_instance.list_cluster_custom_object.return_value = {
+            'items': [mock_ingressroute.to_dict()]
+        }
+
+        get_ingressroutes(mock_api_instance)
+
+        # Assert we call list_cluster_custom_object with expected group
+        mock_api_instance.list_cluster_custom_object.assert_called_with(group='traefik.io',
+                                                                        version='v1alpha1',
+                                                                        plural='ingressroutes')
+
 
 if __name__ == '__main__':
     unittest.main()