Add ArgoCD Integration #85
Conversation
This reverts commit 513ad19.
There was a problem hiding this comment.
Good catch. I'll remove it then
integrations/argocd/CHANGELOG.md
Outdated
|
|
||
| <!-- towncrier release notes start --> | ||
|
|
||
| # Port_Ocean 0.1.0 (2023-08-21) |
There was a problem hiding this comment.
| # Port_Ocean 0.1.0 (2023-08-21) | |
| # 0.1.0 (2023-08-21) |
| """ | ||
| Retrieve ArgoCD resources of a specific type. | ||
|
|
||
| Args: | ||
| resource_type (str): The type of ArgoCD resource to retrieve. | ||
|
|
||
| Returns: | ||
| list[dict[str, Any]]: A list of dictionaries representing the ArgoCD resources | ||
| of the specified type. | ||
| """ |
There was a problem hiding this comment.
I suggest leaving a link to the ArgoCD api reference you relied on, so when other developers jump into this code they will have the full context 😀
There was a problem hiding this comment.
Yeah, I'll add it. You can also find the Swagger API docs here
| applications = await self.get_argocd_resource(resource_type="applications") | ||
| all_deployments = [] | ||
| for application in applications: | ||
| application_metadata = application.get("metadata", {}) |
There was a problem hiding this comment.
Is there any support for pagination from ArgoCD API?
Is it possible for users to have thousands of ArgoCD resources?
There was a problem hiding this comment.
Reading this issue, it does look like the API returns all resources. And yeah, users can have over 1000 resources
There was a problem hiding this comment.
I looked at another issue, and it appears that ArgoCD REST API is yet to support server-side pagination.
| Returns: | ||
| list[dict[str, Any]]: A list of dictionaries representing ArgoCD deployments. | ||
| """ | ||
| applications = await self.get_argocd_resource(resource_type="applications") |
There was a problem hiding this comment.
you can use the class you declared for mapping the Object kinds
There was a problem hiding this comment.
This should be in a separate PR, as well as relevant changelog :)
| ) | ||
|
|
||
| response.raise_for_status() | ||
| return response.json()["items"] |
There was a problem hiding this comment.
I think the logic of parsing the response json shouldn't be in this method but rather in the method calling it/or in the parser, we want this to be agnostic as possible so if we will need to add another api to query we won't need to change this method because the response is different
| return response.json()["items"] | |
| return response.json() |
|
|
||
| # Fetch and update on-call user information for services | ||
| if data_key == "services": | ||
| service_data = await self.update_oncall_users(data, data_key) | ||
| all_data.extend(service_data) | ||
| else: | ||
| all_data.extend(data[data_key]) |
There was a problem hiding this comment.
Why are you doing this?
This is being done on main.py
There was a problem hiding this comment.
My personal repo may have been out of sync with the current. Will update it accordingly
|
Is there any way to subscribe to events in ArgoCD so we can listen to realtime changes? @PeyGis |
|
SonarCloud Quality Gate failed.
|
integrations/argocd/main.py
Outdated
| @ocean.on_resync() | ||
| async def on_resources_resync(kind: str) -> list[dict[Any, Any]]: | ||
| logger.info(f"Listing ArgoCD resource: {kind}") | ||
| argocd_client = init_client() | ||
| return await argocd_client.get_resources(resource_kind=ObjectKind(kind)) |
There was a problem hiding this comment.
what happens when using a non-existing kind?
| class ObjectKind(StrEnum): | ||
| PROJECT = "project" | ||
| APPLICATION = "application" | ||
| CLUSTER = "cluster" |
There was a problem hiding this comment.
no need for a whole file for this
it can be placed in the client file
| def __init__(self, token: str, server_url: str): | ||
| self.token = token | ||
| self.api_url = f"{server_url}/api/v1" | ||
| self.http_client = httpx.AsyncClient(headers=self.api_auth_header) | ||
|
|
||
| @property | ||
| def api_auth_header(self) -> dict[str, Any]: | ||
| return {"Authorization": f"Bearer {self.token}"} |
There was a problem hiding this comment.
| def __init__(self, token: str, server_url: str): | |
| self.token = token | |
| self.api_url = f"{server_url}/api/v1" | |
| self.http_client = httpx.AsyncClient(headers=self.api_auth_header) | |
| @property | |
| def api_auth_header(self) -> dict[str, Any]: | |
| return {"Authorization": f"Bearer {self.token}"} | |
| def __init__(self, token: str, server_url: str): | |
| self.token = token | |
| self.api_url = f"{server_url}/api/v1" | |
| self.api_auth_header = {"Authorization": f"Bearer {self.token}"} | |
| self.http_client = httpx.AsyncClient(headers=self.api_auth_header) |
Description
Added support for ArgoCD projects, applications and cluster resources
What -
Why -
How -
Type of change