Skip to content

ref(schemas): First steps towards isolating query and extension request data #430

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
tkaemming merged 6 commits into from
Aug 21, 2019
Merged

ref(schemas): First steps towards isolating query and extension request data #430

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
71f922c
subschemas
tkaemming Aug 21, 2019
6aec0f3
tidy
tkaemming Aug 21, 2019
f85b70e
freeze Request
tkaemming Aug 21, 2019
188cbac
add type stubs for schema
tkaemming Aug 21, 2019
70b6e8b
don't mutate input value to schema validation
tkaemming Aug 21, 2019
c079c59
Merge branch 'master' into query-schemas
tkaemming Aug 21, 2019
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
13 changes: 9 additions & 4 deletions snuba/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,12 +183,14 @@ def parse_request_body(request):

def validate_request_content(body, schema, timer):
try:
schemas.validate(body, schema)
request = schema.validate(body)
except jsonschema.ValidationError as error:

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I would not explicitly catch jsonschema.ValidationError anymore. In the previous implementation the only method in the try/except block was a direct call to jsonschema validate, thus it was pertinent to expect a jsonschema.ValidationError. Now the type of validation performed by the RequestSchema object is opaque to this method. I would either catch any exception or define a custom validation exception.

@tkaemming tkaemming Aug 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I think this makes sense in general.

It's a bit tricky because the control flow for exceptions changed a bit while you were out with #420. The error payload here is dependent on this being a JSON schema, which I think is appropriate. I'll put some thought into figuring out how this should work. Here's where the error is introspected and relevant payload is generated now:

snuba/snuba/api.py

Lines 77 to 83 in 14ad5d8

elif isinstance(cause, jsonschema.ValidationError):
data = {'error': {
'type': 'schema',
'message': cause.message,
'path': list(cause.path),
'schema': cause.schema,
}}

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

damn, well ok then. I would advise creating a dedicated exception that may wrap the jsonschema.ValidationError but that's not needed in this PR

raise BadRequest(str(error)) from error

timer.mark('validate_schema')

return {**request.body}

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I assume you are creating a new dictionary here in order to have a copy. Is that correct? If yes, I think we should think about a plan for mutations of the body and what we want to allow. My opinion is just returning a frozenset or any immutable mapping and completely disallow mutations of the body of the request. I am suggesting this because we are giving more structure to this code than it was used to have, thus keeping track of mutations to a shared daat structure becomes harder.

@tkaemming tkaemming Aug 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I assume you are creating a new dictionary here in order to have a copy. Is that correct?

Yeah, this is just for change minimization with the existing code.

If yes, I think we should think about a plan for mutations of the body and what we want to allow. My opinion is just returning a frozenset or any immutable mapping and completely disallow mutations of the body of the request. I am suggesting this because we are giving more structure to this code than it was used to have, thus keeping track of mutations to a shared daat structure becomes harder.

I generally agree your viewpoint here. I think typing will help here since we can enforce this by using Mapping rather than MutableMapping. I don't think we should try and address this as part of this change though, but keep it in mind moving forward.

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

mind filing a task to rely on immutable data if you do not plan to address the problem right away ?



@application.route('/query', methods=['GET', 'POST'])
@util.time_request('query')
Expand Down Expand Up @@ -226,7 +228,7 @@ def dataset_query(dataset, body, timer):
assert request.method == 'POST'
ensure_table_exists(dataset)

validate_request_content(body, dataset.get_query_schema(), timer)
body = validate_request_content(body, dataset.get_query_schema(), timer)

result, status = parse_and_run_query(dataset, body, timer)
return (
Expand Down Expand Up @@ -396,8 +398,11 @@ def parse_and_run_query(dataset, body, timer):
@application.route('/internal/sdk-stats', methods=['POST'])
@util.time_request('sdk-stats')
def sdk_distribution(*, timer: Timer):
body = parse_request_body(request)
validate_request_content(body, schemas.SDK_STATS_SCHEMA, timer)
body = validate_request_content(
parse_request_body(request),
schemas.SDK_STATS_SCHEMA,
timer,
)

body['project'] = []
body['aggregations'] = [
Expand Down
225 changes: 157 additions & 68 deletions snuba/schemas.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,50 +1,16 @@
import copy
import itertools
from collections import ChainMap
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Mapping

import jsonschema
import copy


CONDITION_OPERATORS = ['>', '<', '>=', '<=', '=', '!=', 'IN', 'NOT IN', 'IS NULL', 'IS NOT NULL', 'LIKE', 'NOT LIKE']
POSITIVE_OPERATORS = ['>', '<', '>=', '<=', '=', 'IN', 'IS NULL', 'LIKE']


def get_time_series_query_schema_properties(default_granularity: int, default_window: timedelta):
return {
'from_date': {
'type': 'string',
'format': 'date-time',
'default': lambda: (datetime.utcnow().replace(microsecond=0) - default_window).isoformat()
},
'to_date': {
'type': 'string',
'format': 'date-time',
'default': lambda: datetime.utcnow().replace(microsecond=0).isoformat()
},
'granularity': {
'type': 'number',
'default': default_granularity,
},
}


SDK_STATS_SCHEMA = {
'type': 'object',
'properties': {
'groupby': {
'type': 'array',
'items': {
# at the moment the only additional thing you can group by is project_id
'enum': ['project_id']
},
'default': [],
},
**get_time_series_query_schema_properties(
default_granularity=86400, # SDK stats query defaults to 1-day bucketing
default_window=timedelta(days=1),
),
},
'additionalProperties': False,
}

GENERIC_QUERY_SCHEMA = {
'type': 'object',
'properties': {
Expand Down Expand Up @@ -217,46 +183,161 @@ def get_time_series_query_schema_properties(default_granularity: int, default_wi
}
}

EVENTS_QUERY_SCHEMA = {
**copy.deepcopy(GENERIC_QUERY_SCHEMA),
PERFORMANCE_EXTENSION_SCHEMA = {
'type': 'object',
'properties': {
# Never add FINAL to queries, enable sampling
'turbo': {
'type': 'boolean',
'default': False,
},
# Force queries to hit the first shard replica, ensuring the query
# sees data that was written before the query. This burdens the
# first replica, so should only be used when absolutely necessary.
'consistent': {
'type': 'boolean',
'default': False,
},
'debug': {
'type': 'boolean',
'default': False,
},
},
'additionalProperties': False,
}

PROJECT_EXTENSION_SCHEMA = {
'type': 'object',
'properties': {
'project': {
'anyOf': [
{'type': 'number'},
{
'type': 'array',
'items': {'type': 'number'},
'minItems': 1,
},
]
},
},
# Need to select down to the project level for customer isolation and performance
'required': ['project'],
'additionalProperties': False,
}

EVENTS_QUERY_SCHEMA['properties'].update({
**get_time_series_query_schema_properties(

def get_time_series_extension_properties(default_granularity: int, default_window: timedelta):
return {
'type': 'object',
'properties': {
'from_date': {
'type': 'string',
'format': 'date-time',
'default': lambda: (datetime.utcnow().replace(microsecond=0) - default_window).isoformat()
},
'to_date': {
'type': 'string',
'format': 'date-time',
'default': lambda: datetime.utcnow().replace(microsecond=0).isoformat()
},
'granularity': {
'type': 'number',
'default': default_granularity,
},
},
'additionalProperties': False,
}


@dataclass(frozen=True)
class Request:

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

frozen? (provided it allows you to add an additional field in post_init (I never tried)

@tkaemming tkaemming Aug 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This will throw FrozenInstanceError but I can make body a property instead, the ChainMap should be pretty lightweight (compared to a dictionary copy) and is only used once right now anyway.

@tkaemming tkaemming Aug 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Did that with f85b70e.

query: Mapping[str, Any]
extensions: Mapping[str, Mapping[str, Any]]

@property
def body(self):
return ChainMap(self.query, *self.extensions.values())


Schema = Mapping[str, Any] # placeholder for JSON schema


class RequestSchema:
def __init__(self, query_schema: Schema, extensions_schemas: Mapping[str, Schema]):
self.__query_schema = query_schema
self.__extension_schemas = extensions_schemas

self.__composite_schema = {
'type': 'object',
'properties': {},
'required': [],
'definitions': {},
'additionalProperties': False,
}

for schema in itertools.chain([self.__query_schema], self.__extension_schemas.values()):
assert schema['type'] == 'object', 'subschema must be object'
assert schema['additionalProperties'] is False, 'subschema must not allow additional properties'
self.__composite_schema['required'].extend(schema.get('required', []))

for property_name, property_schema in schema['properties'].items():
assert property_name not in self.__composite_schema['properties'], 'subschema cannot redefine property'
self.__composite_schema['properties'][property_name] = property_schema

for definition_name, definition_schema in schema.get('definitions', {}).items():
assert definition_name not in self.__composite_schema['definitions'], 'subschema cannot redefine definition'
self.__composite_schema['definitions'][definition_name] = definition_schema

self.__composite_schema['required'] = set(self.__composite_schema['required'])

def validate(self, value) -> Request:
value = validate_jsonschema(value, self.__composite_schema)

query = {key: value.pop(key) for key in self.__query_schema['properties'].keys() if key in value}

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

What about we supported a query schema like this?

{
    main_field1: value,
    main_field2: value,
    timeseries_extension: {
         time_series_value1: ....
         ...
     },
    performance_extension: {
        ....
       ....
    }
}

While this is not backward compatible, thus we would have to support both formats for a while, this format would have two advantages:

  1. semantics: we would not even have to impose the fields defined by extensions are disjoint. That would be structurally enforced.
  2. The json validation of the extensions would be more sound, since we could just take the extension object and validate it against its extension schema instead of building the composite schema on the fly.

The drawback is that we would commit ourselves to disjoint fields. Which I am not sure it is a bad idea, I cannot think about a strong case for allowing an extension to override a field defined by another one (in terms of validation) that would justify the added complexity of defining the override semantics.

@tkaemming tkaemming Aug 21, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yes, this is my longer term plan, though I would prefer to structure it as:

{
  "query": {…},  # uses generic schema
  "extensions": {
    "name": {…},  # uses an extension schema
    "name": {…}   # uses another extension schema
  }
}

(Note that this is intentionally reflective of the structure of the Request dataclass.)

semantics: we would not even have to impose the fields defined by extensions are disjoint. That would be structurally enforced.

I agree with this.

The json validation of the extensions would be more sound, since we could just take the extension object and validate it against its extension schema instead of building the composite schema on the fly.

I think we'll still want the composite schema (as a concept that validates all subschemas in one pass — not this specific implementation that merges them into top level properties), however, as it makes the validation error easier to understand since it's more reflective of the input in totality instead of independently validating each chunk.

The drawback is that we would commit ourselves to disjoint fields. Which I am not sure it is a bad idea, I cannot think about a strong case for allowing an extension to override a field defined by another one (in terms of validation) that would justify the added complexity of defining the override semantics.

I think disjoint fields are only an artifact of the current API/schema implementation, as soon as we move to a new API/schema implementation and remove the old one, we can remove that requirement since it no longer has any benefit.

In my opinion extensions should be considered totally independent — when they are used in query evaluation, they should only get the generic query data as well as their extension parameters, and not have any visibility in to parameters provided to other extensions. (It'll take a little while to get there but that's at least how I think we should move forward.)

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

ok,


extensions = {}
for extension_name, extension_schema in self.__extension_schemas.items():

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Here we are building subsets of the original value to validate according to the structure of the subschema. Implicitly additionalProperties would be ignored. Why do we need to set additionalProperties = False in the schema ?

@tkaemming tkaemming Aug 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Because right now everything is extracted from a single mapping, it's ambiguous which extension would receive the additional properties as part of its data in the request.extensions map. I don't think discarding them is the right answer either, and disallowing them entirely is just a simpler way of managing that complexity for now. I think we can make it more lax in the future if we move to a more structured query schema similar to the one discussed above.

extensions[extension_name] = {key: value.pop(key) for key in extension_schema['properties'].keys() if key in value}

return Request(query, extensions)


EVENTS_QUERY_SCHEMA = RequestSchema(GENERIC_QUERY_SCHEMA, {

@fpacifici fpacifici Aug 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

What's the advantage of storing all the schemas into one big schemas.py file instead of putting the dataset specific logic (like this class) into the events dataset file? I would put the instantiation of RequestSchema into datasets/events.py like we do for the DDL.

@tkaemming tkaemming Aug 21, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Only change minimization. I think this makes sense as an immediate follow up.

'performance': PERFORMANCE_EXTENSION_SCHEMA,
'project': PROJECT_EXTENSION_SCHEMA,
'timeseries': get_time_series_extension_properties(
default_granularity=3600,
default_window=timedelta(days=5),
),
'project': {
'anyOf': [
{'type': 'number'},
{
'type': 'array',
'items': {'type': 'number'},
'minItems': 1,
})

SDK_STATS_SCHEMA = RequestSchema({
'type': 'object',
'properties': {
'groupby': {
'type': 'array',
'items': {
# at the moment the only additional thing you can group by is project_id
'enum': ['project_id']
},
]
},
# Never add FINAL to queries, enable sampling
'turbo': {
'type': 'boolean',
'default': False,
},
# Force queries to hit the first shard replica, ensuring the query
# sees data that was written before the query. This burdens the
# first replica, so should only be used when absolutely necessary.
'consistent': {
'type': 'boolean',
'default': False,
'default': [],
},
},
'debug': {
'type': 'boolean',
}
'additionalProperties': False,
}, {
'timeseries': get_time_series_extension_properties(
default_granularity=86400, # SDK stats query defaults to 1-day bucketing
default_window=timedelta(days=1),
),
})


def validate(value, schema, set_defaults=True):
def validate_jsonschema(value, schema, set_defaults=True):
"""
Validates a value against the provided schema, returning the validated
value if the value conforms to the schema, otherwise raising a
``jsonschema.ValidationError``.
"""
orig = jsonschema.Draft6Validator.VALIDATORS['properties']

def validate_and_default(validator, properties, instance, schema):
Expand All @@ -270,6 +351,12 @@ def validate_and_default(validator, properties, instance, schema):
for error in orig(validator, properties, instance, schema):
yield error

# Using schema defaults during validation will cause the input value to be
# mutated, so to be on the safe side we create a deep copy of that value to
# avoid unwanted side effects for the calling function.
if set_defaults:
value = copy.deepcopy(value)

validator_cls = jsonschema.validators.extend(
jsonschema.Draft4Validator,
{'properties': validate_and_default}
Expand All @@ -281,6 +368,8 @@ def validate_and_default(validator, properties, instance, schema):
format_checker=jsonschema.FormatChecker()
).validate(value, schema)

return value


def generate(schema):
"""
Expand Down