diff --git a/catalog-rest-service/pom.xml b/catalog-rest-service/pom.xml index 8db148f60dc..9d9d970a9a2 100644 --- a/catalog-rest-service/pom.xml +++ b/catalog-rest-service/pom.xml @@ -247,7 +247,17 @@ org.mockito - mockito-all + mockito-core + test + + + org.mockito + mockito-junit-jupiter + test + + + org.mockito + mockito-inline test diff --git a/catalog-rest-service/src/main/java/org/openmetadata/catalog/airflow/AirflowRESTClient.java b/catalog-rest-service/src/main/java/org/openmetadata/catalog/airflow/AirflowRESTClient.java index 63f80dd3d77..72e7f5ca798 100644 --- a/catalog-rest-service/src/main/java/org/openmetadata/catalog/airflow/AirflowRESTClient.java +++ b/catalog-rest-service/src/main/java/org/openmetadata/catalog/airflow/AirflowRESTClient.java @@ -13,10 +13,13 @@ package org.openmetadata.catalog.airflow; +import com.fasterxml.jackson.core.type.TypeReference; +import java.io.IOException; import java.net.URI; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.util.List; +import java.util.Map; import javax.ws.rs.core.Response; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; @@ -142,19 +145,8 @@ public class AirflowRESTClient extends PipelineServiceClient { @Override public HttpResponse getServiceStatus() { - HttpResponse response; try { - String token = authenticate(); - String authToken = String.format(AUTH_TOKEN, token); - String statusEndPoint = "%s/rest_api/api?api=rest_status"; - String statusUrl = String.format(statusEndPoint, serviceURL); - HttpRequest request = - HttpRequest.newBuilder(URI.create(statusUrl)) - .header(CONTENT_HEADER, CONTENT_TYPE) - .header(AUTH_HEADER, authToken) - .GET() - .build(); - response = client.send(request, HttpResponse.BodyHandlers.ofString()); + HttpResponse response = requestAuthenticatedForJsonContent("%s/rest_api/api?api=rest_status", serviceURL); if (response.statusCode() == 200) { return response; } @@ -180,4 +172,32 @@ public class AirflowRESTClient extends PipelineServiceClient { } throw new PipelineServiceClientException(String.format("Failed to test connection due to %s", response.body())); } + + @Override + public Map getLastIngestionLogs(String pipelineName) { + try { + HttpResponse response = + requestAuthenticatedForJsonContent("%s/rest_api/api?api=last_dag_logs&dag_id=%s", serviceURL, pipelineName); + if (response.statusCode() == 200) { + return JsonUtils.readValue(response.body(), new TypeReference<>() {}); + } + } catch (Exception e) { + throw new PipelineServiceClientException("Failed to get last ingestion logs."); + } + throw new PipelineServiceClientException("Failed to get last ingestion logs."); + } + + private HttpResponse requestAuthenticatedForJsonContent(String stringUrlFormat, Object... stringReplacement) + throws IOException, InterruptedException { + String token = authenticate(); + String authToken = String.format(AUTH_TOKEN, token); + String url = String.format(stringUrlFormat, stringReplacement); + HttpRequest request = + HttpRequest.newBuilder(URI.create(url)) + .header(CONTENT_HEADER, CONTENT_TYPE) + .header(AUTH_HEADER, authToken) + .GET() + .build(); + return client.send(request, HttpResponse.BodyHandlers.ofString()); + } } diff --git a/catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResource.java b/catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResource.java index a6f7bcbba81..0acdef6e9b4 100644 --- a/catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResource.java +++ b/catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResource.java @@ -33,6 +33,7 @@ import java.io.IOException; import java.net.http.HttpResponse; import java.util.List; import java.util.Locale; +import java.util.Map; import javax.json.JsonPatch; import javax.validation.Valid; import javax.validation.constraints.Max; @@ -469,7 +470,7 @@ public class IngestionPipelineResource extends EntityResource lastIngestionLogs = pipelineServiceClient.getLastIngestionLogs(id); + return Response.ok(lastIngestionLogs, MediaType.APPLICATION_JSON_TYPE).build(); + } + private IngestionPipeline getIngestionPipeline(CreateIngestionPipeline create, String user) throws IOException { Source source = buildIngestionSource(create); OpenMetadataServerConnection openMetadataServerConnection = diff --git a/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/JsonUtils.java b/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/JsonUtils.java index 929b8b3a1df..4473a6b58a1 100644 --- a/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/JsonUtils.java +++ b/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/JsonUtils.java @@ -16,6 +16,7 @@ package org.openmetadata.catalog.util; import static org.openmetadata.catalog.util.RestUtil.DATE_TIME_FORMAT; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; @@ -93,6 +94,13 @@ public final class JsonUtils { return OBJECT_MAPPER.readValue(json, clz); } + public static T readValue(String json, TypeReference valueTypeRef) throws IOException { + if (json == null) { + return null; + } + return OBJECT_MAPPER.readValue(json, valueTypeRef); + } + /** Read an array of objects of type {@code T} from json */ public static List readObjects(String json, Class clz) throws IOException { if (json == null) { diff --git a/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/PipelineServiceClient.java b/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/PipelineServiceClient.java index 002d1d2d5bd..95e24daf11d 100644 --- a/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/PipelineServiceClient.java +++ b/catalog-rest-service/src/main/java/org/openmetadata/catalog/util/PipelineServiceClient.java @@ -8,6 +8,7 @@ import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.time.Duration; +import java.util.Map; import org.openmetadata.catalog.api.services.ingestionPipelines.TestServiceConnection; import org.openmetadata.catalog.entity.services.ingestionPipelines.IngestionPipeline; import org.openmetadata.catalog.exception.PipelineServiceClientException; @@ -86,4 +87,7 @@ public abstract class PipelineServiceClient { /* Get the status of a deployed pipeline */ public abstract IngestionPipeline getPipelineStatus(IngestionPipeline ingestionPipeline); + + /* Get the all last run logs of a deployed pipeline */ + public abstract Map getLastIngestionLogs(String pipelineName); } diff --git a/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/AirflowRESTClientIntegrationTest.java b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/AirflowRESTClientIntegrationTest.java new file mode 100644 index 00000000000..5c84f45f0d4 --- /dev/null +++ b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/AirflowRESTClientIntegrationTest.java @@ -0,0 +1,101 @@ +/* + * Copyright 2022 Collate + * Licensed 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. + */ +package org.openmetadata.catalog.airflow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.HashMap; +import java.util.Map; +import lombok.SneakyThrows; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.junit.jupiter.MockitoExtension; +import org.openmetadata.catalog.exception.PipelineServiceClientException; + +@ExtendWith(MockitoExtension.class) +public class AirflowRESTClientIntegrationTest { + + private static final String DAG_NAME = "test_dag"; + private static final String URI_TO_HANDLE_REQUEST = "/"; + + @RegisterExtension private static final HttpServerExtension httpServerExtension = new HttpServerExtension(); + + AirflowRESTClient airflowRESTClient; + + @BeforeEach + void setUp() { + AirflowConfiguration airflowConfiguration = createDefaultAirflowConfiguration(); + airflowRESTClient = new AirflowRESTClient(airflowConfiguration); + httpServerExtension.unregisterHandler(); + } + + @Test + public void testLastIngestionLogsAreRetrievedWhenStatusCodesAre200() { + Map expectedMap = Map.of("key1", "value1", "key2", "value2"); + + registerMockedEndpoints(200, 200); + + assertEquals(expectedMap, airflowRESTClient.getLastIngestionLogs(DAG_NAME)); + } + + @Test + public void testLastIngestionLogsExceptionWhenLoginFails() { + registerMockedEndpoints(404, 200); + + Exception exception = + assertThrows(PipelineServiceClientException.class, () -> airflowRESTClient.getLastIngestionLogs(DAG_NAME)); + + String expectedMessage = "Failed to get last ingestion logs."; + String actualMessage = exception.getMessage(); + + assertEquals(expectedMessage, actualMessage); + } + + @Test + public void testLastIngestionLogsExceptionWhenStatusCode404() { + registerMockedEndpoints(200, 404); + + Exception exception = + assertThrows(PipelineServiceClientException.class, () -> airflowRESTClient.getLastIngestionLogs(DAG_NAME)); + + String expectedMessage = "Failed to get last ingestion logs."; + String actualMessage = exception.getMessage(); + + assertEquals(expectedMessage, actualMessage); + } + + @SneakyThrows + private AirflowConfiguration createDefaultAirflowConfiguration() { + AirflowConfiguration airflowConfiguration = new AirflowConfiguration(); + airflowConfiguration.setApiEndpoint(HttpServerExtension.getUriFor("").toString()); + airflowConfiguration.setUsername("user"); + airflowConfiguration.setPassword("pass"); + airflowConfiguration.setTimeout(60); + return airflowConfiguration; + } + + private void registerMockedEndpoints(int loginStatusCode, int lastDagLogStatusCode) { + String jsonResponse = "{ \"key1\": \"value1\", \"key2\": \"value2\" }"; + + Map pathResponses = new HashMap<>(); + pathResponses.put( + "/rest_api/api?api=last_dag_logs&dag_id=" + DAG_NAME, + new MockResponse(jsonResponse, "application/json", lastDagLogStatusCode)); + pathResponses.put("/api/v1/security/login", new MockResponse("{}", "application/json", loginStatusCode)); + + httpServerExtension.registerHandler(URI_TO_HANDLE_REQUEST, new JsonHandler(pathResponses)); + } +} diff --git a/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/HttpServerExtension.java b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/HttpServerExtension.java new file mode 100644 index 00000000000..b0420b0466f --- /dev/null +++ b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/HttpServerExtension.java @@ -0,0 +1,72 @@ +/* + * Copyright 2022 Collate + * Licensed 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. + */ +package org.openmetadata.catalog.airflow; + +import com.sun.net.httpserver.HttpHandler; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.net.URISyntaxException; +import org.apache.http.client.utils.URIBuilder; +import org.junit.jupiter.api.extension.AfterAllCallback; +import org.junit.jupiter.api.extension.BeforeAllCallback; +import org.junit.jupiter.api.extension.ExtensionContext; + +public class HttpServerExtension implements BeforeAllCallback, AfterAllCallback { + + private static final int PORT; + + static { + try (ServerSocket socket = new ServerSocket(0)) { + socket.setReuseAddress(true); + PORT = socket.getLocalPort(); + } catch (IOException ex) { + throw new RuntimeException("Could not find a free port for testing"); + } + } + + private static final String HOST = "localhost"; + private static final String SCHEME = "http"; + private static final String DEFAULT_CONTEXT = "/"; + + private com.sun.net.httpserver.HttpServer server; + + @Override + public void afterAll(ExtensionContext extensionContext) throws Exception { + if (server != null) { + server.stop(0); + } + } + + @Override + public void beforeAll(ExtensionContext extensionContext) throws Exception { + server = HttpServer.create(new InetSocketAddress(PORT), 0); + server.setExecutor(null); + server.start(); + server.createContext(DEFAULT_CONTEXT); + } + + public static URI getUriFor(String path) throws URISyntaxException { + return new URIBuilder().setScheme(SCHEME).setHost(HOST).setPort(PORT).setPath(path).build(); + } + + public void registerHandler(String uriToHandle, HttpHandler httpHandler) { + server.createContext(uriToHandle, httpHandler); + } + + public void unregisterHandler() { + server.removeContext(DEFAULT_CONTEXT); + } +} diff --git a/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/JsonHandler.java b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/JsonHandler.java new file mode 100644 index 00000000000..4e7643f08c2 --- /dev/null +++ b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/JsonHandler.java @@ -0,0 +1,38 @@ +/* + * Copyright 2022 Collate + * Licensed 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. + */ +package org.openmetadata.catalog.airflow; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpHandler; +import java.io.IOException; +import java.nio.charset.Charset; +import java.util.Map; +import org.apache.commons.io.IOUtils; + +class JsonHandler implements HttpHandler { + + Map pathResponses; + + public JsonHandler(Map pathResponses) { + this.pathResponses = pathResponses; + } + + @Override + public void handle(HttpExchange exchange) throws IOException { + MockResponse response = pathResponses.get(exchange.getRequestURI().toString()); + exchange.getResponseHeaders().add("Content-Type", response.getContentType()); + exchange.sendResponseHeaders(response.getStatusCode(), response.getBody().length()); + IOUtils.write(response.getBody(), exchange.getResponseBody(), Charset.defaultCharset()); + exchange.close(); + } +} diff --git a/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/MockResponse.java b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/MockResponse.java new file mode 100644 index 00000000000..0ed34c7ee6e --- /dev/null +++ b/catalog-rest-service/src/test/java/org/openmetadata/catalog/airflow/MockResponse.java @@ -0,0 +1,25 @@ +/* + * Copyright 2022 Collate + * Licensed 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. + */ +package org.openmetadata.catalog.airflow; + +import lombok.AllArgsConstructor; +import lombok.Getter; + +@Getter +@AllArgsConstructor +public class MockResponse { + + private final String body; + private final String contentType; + private final int statusCode; +} diff --git a/catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResourceUnitTest.java b/catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResourceUnitTest.java new file mode 100644 index 00000000000..097bb2913f6 --- /dev/null +++ b/catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResourceUnitTest.java @@ -0,0 +1,81 @@ +/* + * Copyright 2022 Collate + * Licensed 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. + */ + +package org.openmetadata.catalog.resources.services.ingestionpipelines; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.verify; + +import java.io.IOException; +import java.util.Map; +import javax.ws.rs.core.SecurityContext; +import javax.ws.rs.core.UriInfo; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedConstruction; +import org.mockito.MockedConstruction.Context; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.openmetadata.catalog.CatalogApplicationConfig; +import org.openmetadata.catalog.airflow.AirflowRESTClient; +import org.openmetadata.catalog.jdbi3.CollectionDAO; +import org.openmetadata.catalog.security.Authorizer; +import org.openmetadata.catalog.util.PipelineServiceClient; + +@ExtendWith(MockitoExtension.class) +public class IngestionPipelineResourceUnitTest { + + private static final String DAG_NAME = "test_dag"; + + private IngestionPipelineResource ingestionPipelineResource; + + @Mock UriInfo uriInfo; + + @Mock SecurityContext securityContext; + + @Mock Authorizer authorizer; + + @Mock CollectionDAO collectionDAO; + + @Spy CollectionDAO.IngestionPipelineDAO ingestionPipelineDAO; + + @Mock CatalogApplicationConfig catalogApplicationConfig; + + @BeforeEach + void setUp() { + doReturn(ingestionPipelineDAO).when(collectionDAO).ingestionPipelineDAO(); + ingestionPipelineResource = new IngestionPipelineResource(collectionDAO, authorizer); + } + + @Test + public void testLastIngestionLogsAreRetrievedWhen() throws IOException { + Map expectedMap = Map.of("task", "log"); + try (MockedConstruction mocked = + mockConstruction(AirflowRESTClient.class, this::preparePipelineServiceClient)) { + ingestionPipelineResource.initialize(catalogApplicationConfig); + assertEquals( + expectedMap, ingestionPipelineResource.getLastIngestionLogs(uriInfo, securityContext, DAG_NAME).getEntity()); + PipelineServiceClient client = mocked.constructed().get(0); + verify(client).getLastIngestionLogs(DAG_NAME); + } + } + + private void preparePipelineServiceClient(AirflowRESTClient mockPipelineServiceClient, Context context) { + doReturn(Map.of("task", "log")).when(mockPipelineServiceClient).getLastIngestionLogs(anyString()); + } +} diff --git a/catalog-rest-service/src/test/java/org/openmetadata/catalog/util/JsonUtilsTest.java b/catalog-rest-service/src/test/java/org/openmetadata/catalog/util/JsonUtilsTest.java index 8d6437e274e..3a39f028053 100644 --- a/catalog-rest-service/src/test/java/org/openmetadata/catalog/util/JsonUtilsTest.java +++ b/catalog-rest-service/src/test/java/org/openmetadata/catalog/util/JsonUtilsTest.java @@ -17,7 +17,9 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import com.fasterxml.jackson.core.type.TypeReference; import java.io.IOException; +import java.util.Map; import java.util.UUID; import javax.json.Json; import javax.json.JsonArrayBuilder; @@ -97,4 +99,12 @@ class JsonUtilsTest { assertThrows(JsonException.class, () -> JsonUtils.applyPatch(original, jsonPatchBuilder2.build(), Team.class)); assertTrue(jsonException.getMessage().contains("contains no element for index 3")); } + + @Test + void testReadValuePassingTypeReference() throws IOException { + Map expectedMap = Map.of("key1", "value1", "key2", "value2"); + String json = "{ \"key1\": \"value1\", \"key2\": \"value2\" }"; + TypeReference> mapTypeReference = new TypeReference<>() {}; + assertEquals(expectedMap, JsonUtils.readValue(json, mapTypeReference)); + } } diff --git a/openmetadata-airflow-apis/README.md b/openmetadata-airflow-apis/README.md index 5595aa4c1b8..1fb63a19584 100644 --- a/openmetadata-airflow-apis/README.md +++ b/openmetadata-airflow-apis/README.md @@ -195,7 +195,7 @@ curl -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE - Delete dag based on dag_id. ##### Endpoint: ```text -http://{AIRFLOW_HOST}:{AIRFLOW_PORT}/admin/rest_api/api?api=delete_dag&dag_id=value +http://{AIRFLOW_HOST}:{AIRFLOW_PORT}/rest_api/api?api=delete_dag&dag_id=value ``` ##### Method: - GET @@ -203,7 +203,7 @@ http://{AIRFLOW_HOST}:{AIRFLOW_PORT}/admin/rest_api/api?api=delete_dag&dag_id=va - dag_id - string - The id of dag. ##### Examples: ```bash -curl -X GET http://localhost:8080/admin/rest_api/api?api=delete_dag&dag_id=dag_test +curl -X GET http://localhost:8080/rest_api/api?api=delete_dag&dag_id=dag_test ``` ##### response: ```json diff --git a/openmetadata-airflow-apis/src/openmetadata/api/apis_metadata.py b/openmetadata-airflow-apis/src/openmetadata/api/apis_metadata.py index a9635c993e5..5d9b86bb9ed 100644 --- a/openmetadata-airflow-apis/src/openmetadata/api/apis_metadata.py +++ b/openmetadata-airflow-apis/src/openmetadata/api/apis_metadata.py @@ -79,6 +79,19 @@ APIS_METADATA = [ }, ], }, + { + "name": "last_dag_logs", + "description": "Retrieve all logs from the task instances of a last DAG run", + "http_method": "GET", + "arguments": [ + { + "name": "dag_id", + "description": "The id of the dag", + "form_input_type": "text", + "required": True, + }, + ], + }, { "name": "rest_status", "description": "Get the status of Airflow REST status", diff --git a/openmetadata-airflow-apis/src/openmetadata/api/config.py b/openmetadata-airflow-apis/src/openmetadata/api/config.py index 5655ef61a67..2e1d78baa9f 100644 --- a/openmetadata-airflow-apis/src/openmetadata/api/config.py +++ b/openmetadata-airflow-apis/src/openmetadata/api/config.py @@ -18,7 +18,7 @@ import airflow from airflow import configuration from openmetadata import __version__ -REST_API_ENDPOINT = "/admin/rest_api/api" +REST_API_ENDPOINT = "/rest_api/api" # Getting Versions and Global variables HOSTNAME = socket.gethostname() diff --git a/openmetadata-airflow-apis/src/openmetadata/api/rest_api.py b/openmetadata-airflow-apis/src/openmetadata/api/rest_api.py index d01b1b4bdc7..daa99a7e97c 100644 --- a/openmetadata-airflow-apis/src/openmetadata/api/rest_api.py +++ b/openmetadata-airflow-apis/src/openmetadata/api/rest_api.py @@ -34,6 +34,7 @@ from openmetadata.api.response import ApiResponse from openmetadata.api.utils import jwt_token_secure from openmetadata.operations.delete import delete_dag_id from openmetadata.operations.deploy import DagDeployer +from openmetadata.operations.last_dag_logs import last_dag_logs from openmetadata.operations.status import status from openmetadata.operations.test_connection import test_source_connection from openmetadata.operations.trigger import trigger @@ -130,6 +131,8 @@ class REST_API(AppBuilderBaseView): return self.dag_status() if api == "delete_dag": return self.delete_dag() + if api == "last_dag_logs": + return self.last_dag_logs() raise ValueError( f"Invalid api param {api}. Expected deploy_dag or trigger_dag." @@ -276,3 +279,25 @@ class REST_API(AppBuilderBaseView): status=ApiResponse.STATUS_SERVER_ERROR, error=f"Failed to delete {dag_id} due to {exc} - {traceback.format_exc()}", ) + + def last_dag_logs(self) -> Response: + """ + Retrieve all logs from the task instances of a last DAG run + """ + dag_id: str = self.get_request_arg(request, "dag_id") + + if not dag_id: + return ApiResponse.error( + status=ApiResponse.STATUS_BAD_REQUEST, + error=f"Missing dag_id argument in the request", + ) + + try: + return last_dag_logs(dag_id) + + except Exception as exc: + logging.info(f"Failed to get last run logs for '{dag_id}'") + return ApiResponse.error( + status=ApiResponse.STATUS_SERVER_ERROR, + error=f"Failed to get last run logs for '{dag_id}' due to {exc} - {traceback.format_exc()}", + ) diff --git a/openmetadata-airflow-apis/src/openmetadata/operations/last_dag_logs.py b/openmetadata-airflow-apis/src/openmetadata/operations/last_dag_logs.py new file mode 100644 index 00000000000..287bb1a0c5c --- /dev/null +++ b/openmetadata-airflow-apis/src/openmetadata/operations/last_dag_logs.py @@ -0,0 +1,67 @@ +# Copyright 2022 Collate +# Licensed 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. +""" +Module containing the logic to retrieve all logs from the tasks of a last DAG run +""" +import glob +import os +from pathlib import Path + +from airflow.models import DagModel, DagRun +from flask import Response +from openmetadata.api.response import ApiResponse, ResponseFormat + + +def last_dag_logs(dag_id: str) -> Response: + """ + Validate that the DAG is registered by Airflow and have at least one Run. + If exists, returns all logs for each task instance of the last DAG run. + :param dag_id: DAG to find + :return: API Response + """ + + dag_model = DagModel.get_dagmodel(dag_id=dag_id) + + if not dag_model: + return ApiResponse.not_found(f"DAG '{dag_id}' not found.") + + last_dag_run = dag_model.get_last_dagrun() + + if not last_dag_run: + return ApiResponse.not_found(f"No DAG run found for '{dag_id}'.") + + task_instances = last_dag_run.get_task_instances() + + response = {} + + for task_instance in task_instances: + if os.path.isfile(task_instance.log_filepath): + response[task_instance.task_id] = Path( + task_instance.log_filepath + ).read_text() + # logs could be kept in a directory with the same name than the log file path without extension per attempt + elif os.path.isdir(os.path.splitext(task_instance.log_filepath)[0]): + dir_path = os.path.splitext(task_instance.log_filepath)[0] + sorted_logs = sorted( + filter(os.path.isfile, glob.glob(f"{dir_path}/*.log")), + key=os.path.getmtime, + ) + response[ + task_instance.task_id + ] = f"\n*** Reading local file: {task_instance.log_filepath}\n".join( + [Path(log).read_text() for log in sorted_logs] + ) + else: + return ApiResponse.not_found( + f"Logs for task instance '{task_instance}' of DAG '{dag_id}' not found." + ) + + return ApiResponse.success(response) diff --git a/pom.xml b/pom.xml index df2fa65e273..d140bbfdbc2 100644 --- a/pom.xml +++ b/pom.xml @@ -345,6 +345,18 @@ 2.0.2-beta test + + org.mockito + mockito-inline + ${mockito.version} + test + + + org.mockito + mockito-junit-jupiter + ${mockito.version} + test + com.amazon.redshift redshift-jdbc42