backend/src/appenlight/views/api.py

# -*- coding: utf-8 -*-

# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
#
# 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.

import base64
import io
import datetime
import json
import logging
import urllib.request, urllib.parse, urllib.error
import zlib

from gzip import GzipFile
from pyramid.view import view_config
from pyramid.httpexceptions import HTTPBadRequest

import appenlight.celery.tasks as tasks
from appenlight.lib.api import rate_limiting, check_cors
from appenlight.lib.enums import ParsedSentryEventType
from appenlight.lib.utils import parse_proto
from appenlight.lib.utils.airbrake import parse_airbrake_xml
from appenlight.lib.utils.date_utils import convert_date
from appenlight.lib.utils.sentry import parse_sentry_event
from appenlight.lib.request import JSONException
from appenlight.validators import (
    LogListSchema,
    MetricsListSchema,
    GeneralMetricsListSchema,
    GeneralMetricsPermanentListSchema,
    GeneralMetricSchema,
    GeneralMetricPermanentSchema,
    LogListPermanentSchema,
    ReportListSchema_0_5,
    LogSchema,
    LogSchemaPermanent,
    ReportSchema_0_5,
)

log = logging.getLogger(__name__)


@view_config(
    route_name="api_logs", renderer="string", permission="create", require_csrf=False
)
@view_config(
    route_name="api_log", renderer="string", permission="create", require_csrf=False
)
def logs_create(request):
    """
    Endpoint for log aggregation
    """
    application = request.context.resource
    if request.method.upper() == "OPTIONS":
        return check_cors(request, application)
    else:
        check_cors(request, application, should_return=False)

    params = dict(request.params.copy())
    proto_version = parse_proto(params.get("protocol_version", ""))
    payload = request.unsafe_json_body
    sequence_accepted = request.matched_route.name == "api_logs"

    if sequence_accepted:
        if application.allow_permanent_storage:
            schema = LogListPermanentSchema().bind(utcnow=datetime.datetime.utcnow())
        else:
            schema = LogListSchema().bind(utcnow=datetime.datetime.utcnow())
    else:
        if application.allow_permanent_storage:
            schema = LogSchemaPermanent().bind(utcnow=datetime.datetime.utcnow())
        else:
            schema = LogSchema().bind(utcnow=datetime.datetime.utcnow())

    deserialized_logs = schema.deserialize(payload)
    if sequence_accepted is False:
        deserialized_logs = [deserialized_logs]

    rate_limiting(
        request, application, "per_application_logs_rate_limit", len(deserialized_logs)
    )

    # pprint.pprint(deserialized_logs)

    # we need to split those out so we can process the pkey ones one by one
    non_pkey_logs = [
        log_dict for log_dict in deserialized_logs if not log_dict["primary_key"]
    ]
    pkey_dict = {}
    # try to process the logs as best as we can and group together to reduce
    # the amount of
    for log_dict in deserialized_logs:
        if log_dict["primary_key"]:
            key = (log_dict["primary_key"], log_dict["namespace"])
            if not key in pkey_dict:
                pkey_dict[key] = []
            pkey_dict[key].append(log_dict)

    if non_pkey_logs:
        log.debug("%s non-pkey logs received: %s" % (application, len(non_pkey_logs)))
        tasks.add_logs.delay(application.resource_id, params, non_pkey_logs)
    if pkey_dict:
        logs_to_insert = []
        for primary_key_tuple, payload in pkey_dict.items():
            sorted_logs = sorted(payload, key=lambda x: x["date"])
            logs_to_insert.append(sorted_logs[-1])
        log.debug("%s pkey logs received: %s" % (application, len(logs_to_insert)))
        tasks.add_logs.delay(application.resource_id, params, logs_to_insert)

    log.info(
        "LOG call %s %s client:%s"
        % (application, proto_version, request.headers.get("user_agent"))
    )
    return "OK: Logs accepted"


@view_config(
    route_name="api_request_stats",
    renderer="string",
    permission="create",
    require_csrf=False,
)
@view_config(
    route_name="api_metrics", renderer="string", permission="create", require_csrf=False
)
def request_metrics_create(request):
    """
    Endpoint for performance metrics, aggregates view performance stats
    and converts them to general metric row
    """
    application = request.context.resource
    if request.method.upper() == "OPTIONS":
        return check_cors(request, application)
    else:
        check_cors(request, application, should_return=False)

    params = dict(request.params.copy())
    proto_version = parse_proto(params.get("protocol_version", ""))

    payload = request.unsafe_json_body
    schema = MetricsListSchema()
    dataset = schema.deserialize(payload)

    rate_limiting(
        request, application, "per_application_metrics_rate_limit", len(dataset)
    )

    # looping report data
    metrics = {}
    for metric in dataset:
        server_name = metric.get("server", "").lower() or "unknown"
        start_interval = convert_date(metric["timestamp"])
        start_interval = start_interval.replace(second=0, microsecond=0)

        for view_name, view_metrics in metric["metrics"]:
            key = "%s%s%s" % (metric["server"], start_interval, view_name)
            if start_interval not in metrics:
                metrics[key] = {
                    "requests": 0,
                    "main": 0,
                    "sql": 0,
                    "nosql": 0,
                    "remote": 0,
                    "tmpl": 0,
                    "custom": 0,
                    "sql_calls": 0,
                    "nosql_calls": 0,
                    "remote_calls": 0,
                    "tmpl_calls": 0,
                    "custom_calls": 0,
                    "start_interval": start_interval,
                    "server_name": server_name,
                    "view_name": view_name,
                }
            metrics[key]["requests"] += int(view_metrics["requests"])
            metrics[key]["main"] += round(view_metrics["main"], 5)
            metrics[key]["sql"] += round(view_metrics["sql"], 5)
            metrics[key]["nosql"] += round(view_metrics["nosql"], 5)
            metrics[key]["remote"] += round(view_metrics["remote"], 5)
            metrics[key]["tmpl"] += round(view_metrics["tmpl"], 5)
            metrics[key]["custom"] += round(view_metrics.get("custom", 0.0), 5)
            metrics[key]["sql_calls"] += int(view_metrics.get("sql_calls", 0))
            metrics[key]["nosql_calls"] += int(view_metrics.get("nosql_calls", 0))
            metrics[key]["remote_calls"] += int(view_metrics.get("remote_calls", 0))
            metrics[key]["tmpl_calls"] += int(view_metrics.get("tmpl_calls", 0))
            metrics[key]["custom_calls"] += int(view_metrics.get("custom_calls", 0))

            if not metrics[key]["requests"]:
                # fix this here because validator can't
                metrics[key]["requests"] = 1
                # metrics dict is being built to minimize
                # the amount of queries used
                # in case we get multiple rows from same minute

    normalized_metrics = []
    for metric in metrics.values():
        new_metric = {
            "namespace": "appenlight.request_metric",
            "timestamp": metric.pop("start_interval"),
            "server_name": metric["server_name"],
            "tags": list(metric.items()),
        }
        normalized_metrics.append(new_metric)

    tasks.add_metrics.delay(
        application.resource_id, params, normalized_metrics, proto_version
    )

    log.info(
        "REQUEST METRICS call {} {} client:{}".format(
            application.resource_name, proto_version, request.headers.get("user_agent")
        )
    )
    return "OK: request metrics accepted"


@view_config(
    route_name="api_general_metrics",
    renderer="string",
    permission="create",
    require_csrf=False,
)
@view_config(
    route_name="api_general_metric",
    renderer="string",
    permission="create",
    require_csrf=False,
)
def general_metrics_create(request):
    """
    Endpoint for general metrics aggregation
    """
    application = request.context.resource
    if request.method.upper() == "OPTIONS":
        return check_cors(request, application)
    else:
        check_cors(request, application, should_return=False)

    params = dict(request.params.copy())
    proto_version = parse_proto(params.get("protocol_version", ""))
    payload = request.unsafe_json_body
    sequence_accepted = request.matched_route.name == "api_general_metrics"
    if sequence_accepted:
        if application.allow_permanent_storage:
            schema = GeneralMetricsPermanentListSchema().bind(
                utcnow=datetime.datetime.utcnow()
            )
        else:
            schema = GeneralMetricsListSchema().bind(utcnow=datetime.datetime.utcnow())
    else:
        if application.allow_permanent_storage:
            schema = GeneralMetricPermanentSchema().bind(
                utcnow=datetime.datetime.utcnow()
            )
        else:
            schema = GeneralMetricSchema().bind(utcnow=datetime.datetime.utcnow())

    deserialized_metrics = schema.deserialize(payload)
    if sequence_accepted is False:
        deserialized_metrics = [deserialized_metrics]

    rate_limiting(
        request,
        application,
        "per_application_metrics_rate_limit",
        len(deserialized_metrics),
    )

    tasks.add_metrics.delay(
        application.resource_id, params, deserialized_metrics, proto_version
    )

    log.info(
        "METRICS call {} {} client:{}".format(
            application.resource_name, proto_version, request.headers.get("user_agent")
        )
    )
    return "OK: Metrics accepted"


@view_config(
    route_name="api_reports", renderer="string", permission="create", require_csrf=False
)
@view_config(
    route_name="api_slow_reports",
    renderer="string",
    permission="create",
    require_csrf=False,
)
@view_config(
    route_name="api_report", renderer="string", permission="create", require_csrf=False
)
def reports_create(request):
    """
    Endpoint for exception and slowness reports
    """
    # route_url('reports')
    application = request.context.resource
    if request.method.upper() == "OPTIONS":
        return check_cors(request, application)
    else:
        check_cors(request, application, should_return=False)
    params = dict(request.params.copy())
    proto_version = parse_proto(params.get("protocol_version", ""))
    payload = request.unsafe_json_body
    sequence_accepted = request.matched_route.name == "api_reports"

    if sequence_accepted:
        schema = ReportListSchema_0_5().bind(utcnow=datetime.datetime.utcnow())
    else:
        schema = ReportSchema_0_5().bind(utcnow=datetime.datetime.utcnow())

    deserialized_reports = schema.deserialize(payload)
    if sequence_accepted is False:
        deserialized_reports = [deserialized_reports]
    if deserialized_reports:
        rate_limiting(
            request,
            application,
            "per_application_reports_rate_limit",
            len(deserialized_reports),
        )

        # pprint.pprint(deserialized_reports)
        tasks.add_reports.delay(application.resource_id, params, deserialized_reports)
    log.info(
        "REPORT call  %s, %s client:%s"
        % (application, proto_version, request.headers.get("user_agent"))
    )
    return "OK: Reports accepted"


@view_config(
    route_name="api_airbrake",
    renderer="string",
    permission="create",
    require_csrf=False,
)
def airbrake_xml_compat(request):
    """
    Airbrake compatible endpoint for XML reports
    """
    application = request.context.resource
    if request.method.upper() == "OPTIONS":
        return check_cors(request, application)
    else:
        check_cors(request, application, should_return=False)

    params = dict(request.params.copy())

    error_dict = parse_airbrake_xml(request)
    schema = ReportListSchema_0_5().bind(utcnow=datetime.datetime.utcnow())
    deserialized_reports = schema.deserialize([error_dict])
    rate_limiting(
        request,
        application,
        "per_application_reports_rate_limit",
        len(deserialized_reports),
    )

    tasks.add_reports.delay(application.resource_id, params, deserialized_reports)
    log.info(
        "%s AIRBRAKE call for application %s, api_ver:%s client:%s"
        % (
            500,
            application.resource_name,
            request.params.get("protocol_version", "unknown"),
            request.headers.get("user_agent"),
        )
    )
    return (
        "<notice><id>no-id</id><url>%s</url></notice>"
        % request.registry.settings["mailing.app_url"]
    )


def decompress_gzip(data):
    try:
        fp = io.StringIO(data)
        with GzipFile(fileobj=fp) as f:
            return f.read()
    except Exception as exc:
        raise
        log.error(exc)
        raise HTTPBadRequest()


def decompress_zlib(data):
    try:
        return zlib.decompress(data)
    except Exception as exc:
        raise
        log.error(exc)
        raise HTTPBadRequest()


def decode_b64(data):
    try:
        return base64.b64decode(data)
    except Exception as exc:
        raise
        log.error(exc)
        raise HTTPBadRequest()


@view_config(
    route_name="api_sentry", renderer="string", permission="create", require_csrf=False
)
@view_config(
    route_name="api_sentry_slash",
    renderer="string",
    permission="create",
    require_csrf=False,
)
def sentry_compat(request):
    """
    Sentry compatible endpoint
    """
    application = request.context.resource
    if request.method.upper() == "OPTIONS":
        return check_cors(request, application)
    else:
        check_cors(request, application, should_return=False)

    # handle various report encoding
    content_encoding = request.headers.get("Content-Encoding")
    content_type = request.headers.get("Content-Type")
    if content_encoding == "gzip":
        body = decompress_gzip(request.body)
    elif content_encoding == "deflate":
        body = decompress_zlib(request.body)
    else:
        body = request.body
        # attempt to fix string before decoding for stupid clients
        if content_type == "application/x-www-form-urlencoded":
            body = urllib.parse.unquote(body.decode("utf8"))
        check_char = "{" if isinstance(body, str) else b"{"
        if not body.startswith(check_char):
            try:
                body = decode_b64(body)
                body = decompress_zlib(body)
            except Exception as exc:
                log.info(exc)

    try:
        json_body = json.loads(body.decode("utf8"))
    except ValueError:
        raise JSONException("Incorrect JSON")

    event, event_type = parse_sentry_event(json_body)

    if event_type == ParsedSentryEventType.LOG:
        if application.allow_permanent_storage:
            schema = LogSchemaPermanent().bind(utcnow=datetime.datetime.utcnow())
        else:
            schema = LogSchema().bind(utcnow=datetime.datetime.utcnow())
        deserialized_logs = schema.deserialize(event)
        non_pkey_logs = [deserialized_logs]
        log.debug("%s non-pkey logs received: %s" % (application, len(non_pkey_logs)))
        tasks.add_logs.delay(application.resource_id, {}, non_pkey_logs)
    if event_type == ParsedSentryEventType.ERROR_REPORT:
        schema = ReportSchema_0_5().bind(
            utcnow=datetime.datetime.utcnow(),
            allow_permanent_storage=application.allow_permanent_storage,
        )
        deserialized_reports = [schema.deserialize(event)]
        rate_limiting(
            request,
            application,
            "per_application_reports_rate_limit",
            len(deserialized_reports),
        )
        tasks.add_reports.delay(application.resource_id, {}, deserialized_reports)
    return "OK: Events accepted"