Skip to content

Instantly share code, notes, and snippets.

@kgriffs

kgriffs/datadog_fetch_logs.py

Last active November 27, 2023 17:26
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save kgriffs/e717b8669b9b099b82ac40e11ed25e1a to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save kgriffs/e717b8669b9b099b82ac40e11ed25e1a to your computer and use it in GitHub Desktop.
Download ZIP
DataDog Download Logs Example
Raw
datadog_fetch_logs.py
import json
import requests
import time
class DataDogLogFetcher:
_MAX_LIMIT = 1_000
_FETCH_LIMIT = 500
_SLEEP_SEC = 1
def __init__(self, query, start, end, api_key, app_key):
self._cursor = None
self._query = query
self._api_key = api_key
self._app_key = app_key
self._start_timestamp = start.isoformat()
self._end_timestamp = end.isoformat()
def _next(self):
# Make the API request to get log events
url = f'https://api.datadoghq.com/api/v2/logs/events/search'
headers = {'Content-Type': 'application/json', 'DD-API-KEY': self._api_key, 'DD-APPLICATION-KEY': self._app_key}
params = {
'filter': {'query': query, 'from': self._start_timestamp, 'to': self._end_timestamp},
'page': {'limit': self._FETCH_LIMIT},
}
if self._cursor:
params['page']['cursor'] = self._cursor
response = requests.post(url, headers=headers, json=params)
response.raise_for_status()
result = response.json()
self._cursor = result['meta']['page']['after']
return result['data'] or None
def fetch(self, limit):
all_events = []
while len(all_events) < limit and (batch := self._next()):
all_events += batch
print(f"Fetched a batch of {len(batch)} log events; {len(all_events)} total fetched so far...")
time.sleep(self._SLEEP_SEC)
return all_events
@kgriffs
Copy link
Author

kgriffs commented Nov 21, 2023 •
edited
Loading

Example usage:

end = datetime.now(timezone.utc)
start = end - timedelta(hours=2)
fetcher = DatadogLogFetcher(API_KEY, APP_KEY)

log_events = fetcher.fetch('service:my-app-name', start, end, 10_000)

from collections import defaultdict
import json

def analyze_events(log_events):
    size_by_message = defaultdict(int)
    count_by_message = defaultdict(int)

    for event in log_events:
        attributes = event['attributes']
        size_by_message[attributes['message']] += len(json.dumps(attributes))
        count_by_message[attributes['message']] += 1

    for message, size_bytes in sorted(size_by_message.items(), key=lambda x: x[1], reverse=True):
        size_kb = size_bytes / 1024
        size_mb = size_kb / 1024
        count = count_by_message[message]


        print(f"{size_mb:>8.2f} MiB - {size_kb/count:>6.2f} KiB per Event - {message}")


analyze_events(log_events)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment