"""Script. This script refreshes the cache. """ # flake8: noqa import json import logging import os import re from concurrent.futures import ThreadPoolExecutor import pandas as pd import redis import requests from flask import g from owsrequest import context, flask_request from owsrequest.constants import headers from pandas.io.json import json_normalize from requests.structures import CaseInsensitiveDict from sound_recordings.api import app from sound_recordings.constants import owner_category from sound_recordings.constants.pagination import DEFAULT_LIMIT, DEFAULT_OFFSET from sound_recordings.constants.placements import DEFAULT_MIN_FOLLOWERS from sound_recordings.logic import ( demographics, downloads, placements, playlists, sound_recording_metadata, source_of_streams, streams, streams_breakdown, top_countries, top_markets, top_sound_recordings, ) logging.getLogger("werkzeug").setLevel(logging.ERROR) logging.getLogger("requests").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) logging.getLogger("botocore").setLevel(logging.WARNING) logging.getLogger("boto3").setLevel(logging.WARNING) DD_API_KEY = os.environ.get("DATADOG_API_KEY") DD_APP_KEY = os.environ.get("DATADOG_APP_KEY") NUM_WORKERS = os.environ.get("NUM_WORKERS", 5) NUM_HITS = os.environ.get("NUM_HITS", 1000) dd_headers = { "content-type": "application/json", "DD-API-KEY": DD_API_KEY, "DD-APPLICATION-KEY": DD_APP_KEY, } start_date = None end_date = None countries = [] store_ids = [] alt_limit = 25 request_counter = 0 env_namespace = os.environ.get("Environment") redis_client = redis.StrictRedis(host=os.environ.get("REDIS_HOST"), port=6379, db=0) query_now = { "query": """ service:ows-analytics-sr environment:{} @http.method:GET @level_name:INFO -@http.url_details.path:( \/sound-recording\/*\/metadata OR \/feeds OR *podcast* OR \/network-daily-downloads OR \/episode-daily-downloads OR \/account) """.format( env_namespace ), "time": {"from": "now - 24h", "to": "now"}, "sort": "desc", "limit": 1000, } query_old = { "query": """ service:ows-analytics-sr environment:{} @http.method:GET @level_name:INFO -@http.url_details.path:( \/sound-recording\/*\/metadata OR \/feeds OR *podcast* OR \/network-daily-downloads OR \/episode-daily-downloads OR \/account) """.format( env_namespace ), "time": {"from": "now - 48h", "to": "now - 24h"}, "sort": "desc", "limit": 1000, } function_dict = { "metadata": { "func": sound_recording_metadata.get_sound_recording_metadata, "args": {"include_deleted": False}, }, "placements": { "func": placements.get_placements, "args": { "store_ids": list(map(lambda s: s["id"], placements.SOURCES)), "owner_categories": owner_category.DEFAULTS, "min_followers": DEFAULT_MIN_FOLLOWERS, }, }, "source-of-streams": { "func": source_of_streams.get_source_breakdown, "args": { "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": ["theorchard"], }, }, "streams-breakdown": { "func": streams_breakdown.get_streams_breakdown, "args": { "distributors": ["theorchard"], "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, }, }, "streams": { "func": streams.get_streams_all, "args": { "distributors": ["theorchard"], "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, }, }, "downloads": { "func": downloads.get_downloads, "args": { "distributors": ["theorchard"], "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, }, }, "top-markets": { "func": top_markets.get_top_markets, "args": { "distributors": ["theorchard"], }, }, "playlists": { "func": playlists.get_playlists, "args": { "distributors": ["theorchard"], "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "offset": DEFAULT_OFFSET, "limit": DEFAULT_LIMIT, }, }, "top-countries-streams": { "func": top_countries.get_top_countries_streams, "args": { "distributors": ["theorchard"], "store_ids": store_ids, "start_date": start_date, "end_date": end_date, }, }, "top-countries-downloads": { "func": top_countries.get_top_countries_downloads, "args": { "distributors": ["theorchard"], "store_ids": store_ids, "start_date": start_date, "end_date": end_date, }, }, "demographics": { "func": demographics.get_demographics, "args": { "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": ["theorchard"], }, }, "top-sound-recordings": { "func": top_sound_recordings.get_top_sound_recordings, "args": {"distributors": ["theorchard"], "limit": alt_limit, "offset": 0}, }, "recent-placements": { "func": placements.get_recent_placements, "args": { "limit": alt_limit, "offset": 0, "store_ids": list(map(lambda s: s["id"], placements.SOURCES)), "owner_categories": owner_category.DEFAULTS, "min_followers": DEFAULT_MIN_FOLLOWERS, }, }, } def _fetch_next_log_page(next_id, query): """ Fetch a page of logs (max 1000 records). Returns: list: datadog log dicts """ if isinstance(next_id, str): query["startAt"] = next_id logs = requests.post( "https://api.datadoghq.com/api/v1/logs-queries/list", headers=dd_headers, data=json.dumps(query), ) return json.loads(logs.text) def fetch_dd_log_list(query): """ Fetch the last days worth of logs from datadog. Returns: list: datadog log dicts """ log_list = [] next_id = True while next_id is not None: logs_dict = _fetch_next_log_page(next_id, query) log_list += logs_dict["logs"] next_id = logs_dict["nextLogId"] print("retrieved {} logs".format(len(log_list))) return log_list def add_log_context(logs): """ Add request_context to logs. Returns: list: datadog log dicts """ for item in logs: try: request_context = json.dumps( item["content"]["attributes"]["resources"], sort_keys=True ) item.update({"request_context": request_context}) except KeyError: # skip logs without resources information pass return logs def get_top_hits(logs, num_of_hits): """ Parse logs and get the most hit endpoints. Returns: list: dicts of endpoint, context and hits """ logframe = pd.DataFrame(json_normalize(logs)) url_hits = logframe.groupby( ["content.attributes.http.url_details.path", "request_context"] ).size() top_hits = url_hits.nlargest(num_of_hits).reset_index(name="top_hits") print("total hits: ", len(logs)) print("unique hits: ", len(url_hits)) print("cache coverage: ", top_hits["top_hits"].sum()) return top_hits.to_dict("records") def _get_endpoint_from_url(log_item): """ Search through endpoints and return related function. Resolves: /top-sound-recordings /top-metrics /recent-placements Returns: list: dicts of endpoint, context and hits """ url = log_item["content.attributes.http.url_details.path"] raw_headers = json.loads(log_item["request_context"]) fixed_headers = { headers.GRASS_ACCOUNT_TYPE: raw_headers.get("account_type", None), headers.GRASS_ACCOUNT_ID: raw_headers.get("account_id", None), headers.ORCHARD_USER_ID: raw_headers.get("user_id", None), headers.ORCHARD_PROFILE_TYPE: raw_headers.get("profile_type", None), headers.ORCHARD_PROFILE_ID: raw_headers.get("profile_id", None), headers.ORCHARD_IDENTITY_ID: raw_headers.get("identity_id", None), headers.ORCHARD_ROLES: ",".join(raw_headers.get("roles", "")), "correlation-id": "refresh_cache", "Correlation-Id": "refresh_cache", } request_context = context.get_request_context_from_headers( CaseInsensitiveDict(fixed_headers), label_profile=True ) with app.app_context(): g.request_context = request_context g.ows = flask_request.get_ows() g.ows.correlation_id = "refresh_cache" with app.test_request_context(headers=fixed_headers): if url == "/top-sound-recordings": func = function_dict["top-sound-recordings"]["func"] kwargs = function_dict["top-sound-recordings"]["args"] func(request_context, **kwargs) elif url == "/top-metrics": func = function_dict["top-metrics"]["func"] kwargs = function_dict["top-metrics"]["args"] func(request_context, **kwargs) elif url == "/recent-placements": func = function_dict["recent-placements"]["func"] kwargs = function_dict["recent-placements"]["args"] func(request_context, **kwargs) elif re.search("\/sound-recording\/", url) is not None: _get_sound_recording_endpoint(log_item, request_context) else: raise Exception("Unknown URL {}".format(url)) return def _get_sound_recording_endpoint(log_item, request_context): """ Search through endpoints and return related function. Relevant Endpoints: /sound-recording/ + /metadata /placements /source-of-streams /streams-breakdown /streams /downloads /top-markets /playlists /top-countries-streams /top-countries-downloads /demographics Returns: list: dicts of endpoint, context and hits """ fields = log_item["content.attributes.http.url_details.path"].split("/") isrc = fields[2] func = function_dict[fields[3]]["func"] kwargs = function_dict[fields[3]]["args"] return func(request_context, isrc, **kwargs) def cycle_through_endpoints(logs): """Go through all endpoints and get each request.""" with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor: results = executor.map(_get_endpoint_from_url, logs) for r in results: global request_counter request_counter += 1 if request_counter % 10 == 0: print("done, ", request_counter) return True def calc_coverage(old_raw_logs, new_raw_logs): """Calculate coverage from prewarming.""" old_logs = get_top_hits(old_raw_logs, NUM_HITS) new_logs = get_top_hits(new_raw_logs, 10000) url_prop = "content.attributes.http.url_details.path" request_prop = "request_context" new_log_keys = [i[url_prop] + i[request_prop] for i in new_logs] new_log_dict = dict(zip(new_log_keys, new_logs)) total_sum = 0 unique_sum = 0 for old_log in old_logs: key = old_log[url_prop] + old_log[request_prop] new_item = new_log_dict.get(key, {}) new_count = new_item.get("top_hits", 0) total_sum += new_count if new_count > 0: unique_sum += 1 return total_sum, unique_sum def main(): """Execute script.""" # Get the last 24h of logs new_logs = fetch_dd_log_list(query_now) new_logs = add_log_context(new_logs) # Get the last 48-24h of logs old_logs = fetch_dd_log_list(query_old) old_logs = add_log_context(old_logs) top_new_hits = get_top_hits(new_logs, NUM_HITS) total_coverage, unique_coverage = calc_coverage(old_logs, new_logs) print("Yesterday, {}/{} hits were used".format(unique_coverage, NUM_HITS)) print("This was, {}/{} of requests".format(total_coverage, len(new_logs))) redis_client.flushall() cycle_through_endpoints(top_new_hits)