You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jscheffl (via GitHub)" <gi...@apache.org> on 2024/11/27 22:53:11 UTC
[PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
jscheffl opened a new pull request, #44434:
URL: https://github.com/apache/airflow/pull/44434
After PR https://github.com/apache/airflow/pull/44311, https://github.com/apache/airflow/pull/44330 and #44433 this PR is a cleanup and removes the old Internal API extensions.
For review: Only the last commit is relevant, the other 9 are from referenced base PRs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #44434:
URL: https://github.com/apache/airflow/pull/44434#discussion_r1864843918
##########
providers/src/airflow/providers/edge/CHANGELOG.rst:
##########
@@ -27,6 +27,22 @@
Changelog
---------
+0.9.0pre0
Review Comment:
It is an "internal breaking change" to remove these bindings. So I thought it is valid to get to a 0.9.0... but don't fear this means we are close to 1.0... still I assume there will be a 0.10.0 and 0.11.0 probably :-D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #44434:
URL: https://github.com/apache/airflow/pull/44434#discussion_r1864597238
##########
providers/src/airflow/providers/edge/CHANGELOG.rst:
##########
@@ -27,6 +27,22 @@
Changelog
---------
+0.9.0pre0
Review Comment:
Aren't we supposed to merge these two ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #44434:
URL: https://github.com/apache/airflow/pull/44434#issuecomment-2509704262
> There is some devilish trickery with this one... Seems like it's no change at all:
> But ... It looks good:
Now after Part 3 is merged looks much better:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #44434:
URL: https://github.com/apache/airflow/pull/44434#discussion_r1864456760
##########
providers/src/airflow/providers/edge/worker_api/routes/_v2_routes.py:
##########
@@ -20,12 +20,15 @@
import json
import logging
-from functools import cache
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
from uuid import uuid4
from flask import Response, request
+from airflow.api_internal.endpoints.rpc_api_endpoint import ( # type: ignore[attr-defined]
+ # Note: This is just for compatibility with Airflow 2.10, not working for Airflow 3 / main as removed
+ initialize_method_map,
Review Comment:
@potiuk FYI in my cleanup I realized... as the internal API is "gone" (virtually, at least not be usable, will contribute in a moment to clean as well...) and as workaround a copy was made... actually that part of code is only working in Airflow 2.10. Therefore removing the copy of the full method map here in the cleanup and sourcing this from the core (=2.10)
This way the "traces" are reduced, just a import from the past is used until we drop 2.10 support in Edge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #44434:
URL: https://github.com/apache/airflow/pull/44434
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #44434:
URL: https://github.com/apache/airflow/pull/44434#discussion_r1861323456
##########
providers/src/airflow/providers/edge/cli/api_client.py:
##########
@@ -0,0 +1,169 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime
+from http import HTTPStatus
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+from urllib.parse import quote, urljoin, urlparse
+
+import requests
+import tenacity
+from requests.exceptions import ConnectionError
+from urllib3.exceptions import NewConnectionError
+
+from airflow.configuration import conf
+from airflow.exceptions import AirflowException
+from airflow.providers.edge.worker_api.auth import jwt_signer
+from airflow.providers.edge.worker_api.datamodels import (
+ EdgeJobFetched,
+ PushLogsBody,
+ WorkerQueuesBody,
+ WorkerStateBody,
+)
+from airflow.utils.state import TaskInstanceState # noqa: TC001
+
+if TYPE_CHECKING:
+ from airflow.models.taskinstancekey import TaskInstanceKey
+ from airflow.providers.edge.models.edge_worker import EdgeWorkerState
+
+logger = logging.getLogger(__name__)
+
+
+class AirflowHttpException(AirflowException):
+ """Raise when there is a problem during an http request on the internal API decorator."""
+
+ def __init__(self, message: str, status_code: HTTPStatus):
+ super().__init__(message)
+ self.status_code = status_code
+
+
+def _is_retryable_exception(exception: BaseException) -> bool:
+ """
+ Evaluate which exception types to retry.
+
+ This is especially demanded for cases where an application gateway or Kubernetes ingress can
+ not find a running instance of a webserver hosting the API (HTTP 502+504) or when the
+ HTTP request fails in general on network level.
+
+ Note that we want to fail on other general errors on the webserver not to send bad requests in an endless loop.
+ """
+ retryable_status_codes = (HTTPStatus.BAD_GATEWAY, HTTPStatus.GATEWAY_TIMEOUT)
+ return (
+ isinstance(exception, AirflowHttpException)
+ and exception.status_code in retryable_status_codes
+ or isinstance(exception, (ConnectionError, NewConnectionError))
+ )
+
+
+@tenacity.retry(
+ stop=tenacity.stop_after_attempt(10),
+ wait=tenacity.wait_exponential(min=1),
+ retry=tenacity.retry_if_exception(_is_retryable_exception),
+ before_sleep=tenacity.before_log(logger, logging.WARNING),
+)
+def _make_generic_request(method: str, rest_path: str, data: str | None = None) -> Any:
+ signer = jwt_signer()
+ api_url = conf.get("edge", "api_url")
+ path = urlparse(api_url).path.replace("/rpcapi", "")
+ headers = {
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ "Authorization": signer.generate_signed_token({"method": str(Path(path, rest_path))}),
+ }
+ api_endpoint = urljoin(api_url, rest_path)
+ response = requests.request(method, url=api_endpoint, data=data, headers=headers)
+ if response.status_code == 204:
+ return None
+ if response.status_code != 200:
+ raise AirflowHttpException(
+ f"Got {response.status_code}:{response.reason} when sending "
+ f"the internal api request: {response.text}",
+ HTTPStatus(response.status_code),
+ )
+ return json.loads(response.content)
+
+
+def worker_register(
+ hostname: str, state: EdgeWorkerState, queues: list[str] | None, sysinfo: dict
+) -> datetime:
+ """Register worker with the Edge API."""
+ result = _make_generic_request(
+ "POST",
+ f"worker/{quote(hostname)}",
+ WorkerStateBody(state=state, jobs_active=0, queues=queues, sysinfo=sysinfo).model_dump_json(),
+ )
+ return datetime.fromisoformat(result)
+
+
+def worker_set_state(
Review Comment:
The docstring for the `worker_set_state` function is misleading. It should be updated to reflect that this function updates the worker state, not registers it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate Edge calls for Worker to FastAPI part 4 - Cleanup [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #44434:
URL: https://github.com/apache/airflow/pull/44434#issuecomment-2509747746
> Now after Part 3 is merged looks much better:
:)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org