Skip to content

fix(ingest): better warnings and error handling for rest sink #2800

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shirshanka merged 1 commit into from
Jun 30, 2021
Merged

fix(ingest): better warnings and error handling for rest sink #2800

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions metadata-ingestion/src/datahub/emitter/rest_emitter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import logging
import shlex
from collections import OrderedDict
from json.decoder import JSONDecodeError
from typing import Any, List, Optional, Union

import requests
Expand Down Expand Up @@ -63,6 +64,10 @@ class DatahubRestEmitter:
_session: requests.Session

def __init__(self, gms_server: str, token: Optional[str] = None):
if ":9002" in gms_server:
Copy link
Contributor

@shirshanka shirshanka Jun 30, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can this be better detected by hitting some diagnostic endpoint where the server actually responds with "what" it is?

Copy link
Contributor Author

@hsheth2 hsheth2 Jul 1, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yep - added that to my TODO list

logger.warn(
"the rest emitter should connect to GMS (usually port 8080) instead of frontend"
)
self._gms_server = gms_server
self._token = token

Expand Down Expand Up @@ -119,10 +124,16 @@ def _emit_generic(self, url: str, payload: str) -> None:

response.raise_for_status()
except HTTPError as e:
info = response.json()
raise OperationalError(
"Unable to emit metadata to DataHub GMS", info
) from e
try:
info = response.json()
raise OperationalError(
"Unable to emit metadata to DataHub GMS", info
) from e
except JSONDecodeError:
# If we can't parse the JSON, just raise the original error.
raise OperationalError(
"Unable to emit metadata to DataHub GMS", {"message": str(e)}
) from e
except RequestException as e:
raise OperationalError(
"Unable to emit metadata to DataHub GMS", {"message": str(e)}
Expand Down