From adb902344f8d4076721cc4603046fd5d4688fb28 Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 30 Mar 2022 06:52:14 -0400 Subject: [PATCH 01/10] added redis folder to fork redis --- filibuster/instrumentation/redis/__init__.py | 224 +++++++++++++++++++ 1 file changed, 224 insertions(+) create mode 100644 filibuster/instrumentation/redis/__init__.py diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py new file mode 100644 index 0000000..449b304 --- /dev/null +++ b/filibuster/instrumentation/redis/__init__.py @@ -0,0 +1,224 @@ +# Copyright The OpenTelemetry 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. +# +""" +Instrument `redis`_ to report Redis queries. + +There are two options for instrumenting code. The first option is to use the +``opentelemetry-instrumentation`` executable which will automatically +instrument your Redis client. The second is to programmatically enable +instrumentation via the following code: + +.. _redis: https://pypi.org/project/redis/ + +Usage +----- + +.. code:: python + + from opentelemetry.instrumentation.redis import RedisInstrumentor + import redis + + + # Instrument redis + RedisInstrumentor().instrument() + + # This will report a span with the default settings + client = redis.StrictRedis(host="localhost", port=6379) + client.get("my-key") + +The `instrument` method accepts the following keyword args: + +tracer_provider (TracerProvider) - an optional tracer provider + +request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request +this function signature is: def request_hook(span: Span, instance: redis.connection.Connection, args, kwargs) -> None + +response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request +this function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None + +for example: + +.. code: python + + from opentelemetry.instrumentation.redis import RedisInstrumentor + import redis + + def request_hook(span, instance, args, kwargs): + if span and span.is_recording(): + span.set_attribute("custom_user_attribute_from_request_hook", "some-value") + + def response_hook(span, instance, response): + if span and span.is_recording(): + span.set_attribute("custom_user_attribute_from_response_hook", "some-value") + + # Instrument redis with hooks + RedisInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook) + + # This will report a span with the default settings and the custom attributes added from the hooks + client = redis.StrictRedis(host="localhost", port=6379) + client.get("my-key") + +API +--- +""" +import typing +from typing import Any, Collection + +import redis +from wrapt import wrap_function_wrapper + +from opentelemetry import trace +from opentelemetry.instrumentation.instrumentor import BaseInstrumentor +from opentelemetry.instrumentation.redis.package import _instruments +from opentelemetry.instrumentation.redis.util import ( + _extract_conn_attributes, + _format_command_args, +) +from opentelemetry.instrumentation.redis.version import __version__ +from opentelemetry.instrumentation.utils import unwrap +from opentelemetry.semconv.trace import SpanAttributes +from opentelemetry.trace import Span + +_DEFAULT_SERVICE = "redis" + +_RequestHookT = typing.Optional[ + typing.Callable[ + [Span, redis.connection.Connection, typing.List, typing.Dict], None + ] +] +_ResponseHookT = typing.Optional[ + typing.Callable[[Span, redis.connection.Connection, Any], None] +] + + +def _set_connection_attributes(span, conn): + if not span.is_recording(): + return + for key, value in _extract_conn_attributes( + conn.connection_pool.connection_kwargs + ).items(): + span.set_attribute(key, value) + + +def _instrument( + tracer, + request_hook: _RequestHookT = None, + response_hook: _ResponseHookT = None, +): + def _traced_execute_command(func, instance, args, kwargs): + query = _format_command_args(args) + name = "" + if len(args) > 0 and args[0]: + name = args[0] + else: + name = instance.connection_pool.connection_kwargs.get("db", 0) + with tracer.start_as_current_span( + name, kind=trace.SpanKind.CLIENT + ) as span: + if span.is_recording(): + span.set_attribute(SpanAttributes.DB_STATEMENT, query) + _set_connection_attributes(span, instance) + span.set_attribute("db.redis.args_length", len(args)) + if callable(request_hook): + request_hook(span, instance, args, kwargs) + response = func(*args, **kwargs) + if callable(response_hook): + response_hook(span, instance, response) + return response + + def _traced_execute_pipeline(func, instance, args, kwargs): + cmds = [_format_command_args(c) for c, _ in instance.command_stack] + resource = "\n".join(cmds) + + span_name = " ".join([args[0] for args, _ in instance.command_stack]) + + with tracer.start_as_current_span( + span_name, kind=trace.SpanKind.CLIENT + ) as span: + if span.is_recording(): + span.set_attribute(SpanAttributes.DB_STATEMENT, resource) + _set_connection_attributes(span, instance) + span.set_attribute( + "db.redis.pipeline_length", len(instance.command_stack) + ) + response = func(*args, **kwargs) + if callable(response_hook): + response_hook(span, instance, response) + return response + + pipeline_class = ( + "BasePipeline" if redis.VERSION < (3, 0, 0) else "Pipeline" + ) + redis_class = "StrictRedis" if redis.VERSION < (3, 0, 0) else "Redis" + + wrap_function_wrapper( + "redis", f"{redis_class}.execute_command", _traced_execute_command + ) + wrap_function_wrapper( + "redis.client", + f"{pipeline_class}.execute", + _traced_execute_pipeline, + ) + wrap_function_wrapper( + "redis.client", + f"{pipeline_class}.immediate_execute_command", + _traced_execute_command, + ) + + +class RedisInstrumentor(BaseInstrumentor): + """An instrumentor for Redis + See `BaseInstrumentor` + """ + + def instrumentation_dependencies(self) -> Collection[str]: + return _instruments + + def _instrument(self, **kwargs): + """Instruments the redis module + + Args: + **kwargs: Optional arguments + ``tracer_provider``: a TracerProvider, defaults to global. + ``response_hook``: An optional callback which is invoked right before the span is finished processing a response. + """ + tracer_provider = kwargs.get("tracer_provider") + tracer = trace.get_tracer( + __name__, __version__, tracer_provider=tracer_provider + ) + _instrument( + tracer, + request_hook=kwargs.get("request_hook"), + response_hook=kwargs.get("response_hook"), + ) + + def _uninstrument(self, **kwargs): + if redis.VERSION < (3, 0, 0): + unwrap(redis.StrictRedis, "execute_command") + unwrap(redis.StrictRedis, "pipeline") + unwrap(redis.Redis, "pipeline") + unwrap( + redis.client.BasePipeline, # pylint:disable=no-member + "execute", + ) + unwrap( + redis.client.BasePipeline, # pylint:disable=no-member + "immediate_execute_command", + ) + else: + unwrap(redis.Redis, "execute_command") + unwrap(redis.Redis, "pipeline") + unwrap(redis.client.Pipeline, "execute") + unwrap(redis.client.Pipeline, "immediate_execute_command") \ No newline at end of file From f2a8a9da63b91632d887048e0177731dddb48d34 Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 30 Mar 2022 07:00:54 -0400 Subject: [PATCH 02/10] fixed failing _instrumentor code --- filibuster/instrumentation/redis/__init__.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index 449b304..6d34325 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -81,7 +81,6 @@ def response_hook(span, instance, response): from opentelemetry import trace from opentelemetry.instrumentation.instrumentor import BaseInstrumentor -from opentelemetry.instrumentation.redis.package import _instruments from opentelemetry.instrumentation.redis.util import ( _extract_conn_attributes, _format_command_args, @@ -183,9 +182,6 @@ class RedisInstrumentor(BaseInstrumentor): See `BaseInstrumentor` """ - def instrumentation_dependencies(self) -> Collection[str]: - return _instruments - def _instrument(self, **kwargs): """Instruments the redis module From ce1b6cc71933f19054df16e1e8a9cb65fb266b5e Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 13 Apr 2022 01:29:00 -0400 Subject: [PATCH 03/10] added execution index to redis --- filibuster/instrumentation/redis/__init__.py | 171 ++++++++++++++++++- 1 file changed, 168 insertions(+), 3 deletions(-) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index 6d34325..459e260 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -30,7 +30,6 @@ from opentelemetry.instrumentation.redis import RedisInstrumentor import redis - # Instrument redis RedisInstrumentor().instrument() @@ -79,6 +78,18 @@ def response_hook(span, instance, response): import redis from wrapt import wrap_function_wrapper +import functools +import types +import json + +import sys +import re +import hashlib +import os + +from threading import Lock + +from opentelemetry import context from opentelemetry import trace from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.redis.util import ( @@ -90,7 +101,58 @@ def response_hook(span, instance, response): from opentelemetry.semconv.trace import SpanAttributes from opentelemetry.trace import Span -_DEFAULT_SERVICE = "redis" +from filibuster.global_context import get_value as _filibuster_global_context_get_value +from filibuster.global_context import set_value as _filibuster_global_context_set_value +from filibuster.execution_index import execution_index_new, execution_index_fromstring, \ + execution_index_tostring, execution_index_push, execution_index_pop +from filibuster.instrumentation.helpers import get_full_traceback_hash, should_load_counterexample_file, \ + counterexample_file +from filibuster.logger import warning, debug, notice, info +from filibuster.vclock import vclock_new, vclock_tostring, vclock_fromstring, vclock_increment, vclock_merge +from filibuster.nginx_http_special_response import get_response +from filibuster.server_helpers import should_fail_request_with, load_counterexample +from filibuster.datatypes import TestExecution +from filibuster.instrumentation.helpers import get_full_traceback_hash + +# A key to a context variable to avoid creating duplicate spans when instrumenting +# both, Session.request and Session.send, since Session.request calls into Session.send +_FILIBUSTER_SUPPRESS_REQUESTS_INSTRUMENTATION_KEY = "filibuster_suppress_requests_instrumentation" + +# We do not want to instrument the instrumentation, so use this key to detect when we +# are inside of a Filibuster instrumentation call to suppress further instrumentation. +_FILIBUSTER_INSTRUMENTATION_KEY = "filibuster_instrumentation" + +# Key for the Filibuster vclock in the context. +_FILIBUSTER_VCLOCK_KEY = "filibuster_vclock" + +# Key for the Filibuster origin vclock in the context. +_FILIBUSTER_ORIGIN_VCLOCK_KEY = "filibuster_origin_vclock" + +# Key for the Filibuster execution index in the context. +_FILIBUSTER_EXECUTION_INDEX_KEY = "filibuster_execution_index" + +# Key for the Filibuster request id in the context. +_FILIBUSTER_REQUEST_ID_KEY = "filibuster_request_id" + +# Key for Filibuster vclock mapping. +_FILIBUSTER_VCLOCK_BY_REQUEST_KEY = "filibuster_vclock_by_request" +_filibuster_global_context_set_value(_FILIBUSTER_VCLOCK_BY_REQUEST_KEY, {}) + +# Mutex for vclock and execution_index. +ei_and_vclock_mutex = Lock() + +# Last used execution index. +# (this is mutated under the same mutex as the vclock.) +_FILIBUSTER_EI_BY_REQUEST_KEY = "filibuster_execution_indices_by_request" +_filibuster_global_context_set_value(_FILIBUSTER_EI_BY_REQUEST_KEY, {}) + +if should_load_counterexample_file(): + notice("Counterexample file present!") + counterexample = load_counterexample(counterexample_file()) + counterexample_test_execution = TestExecution.from_json(counterexample['TestExecution']) if counterexample else None + print(counterexample_test_execution.failures) +else: + counterexample = None _RequestHookT = typing.Optional[ typing.Callable[ @@ -115,6 +177,7 @@ def _instrument( tracer, request_hook: _RequestHookT = None, response_hook: _ResponseHookT = None, + service_name=None ): def _traced_execute_command(func, instance, args, kwargs): query = _format_command_args(args) @@ -123,6 +186,7 @@ def _traced_execute_command(func, instance, args, kwargs): name = args[0] else: name = instance.connection_pool.connection_kwargs.get("db", 0) + with tracer.start_as_current_span( name, kind=trace.SpanKind.CLIENT ) as span: @@ -130,9 +194,11 @@ def _traced_execute_command(func, instance, args, kwargs): span.set_attribute(SpanAttributes.DB_STATEMENT, query) _set_connection_attributes(span, instance) span.set_attribute("db.redis.args_length", len(args)) + if callable(request_hook): request_hook(span, instance, args, kwargs) - response = func(*args, **kwargs) + + response = _instrumented_redis_call(service_name, func, name, args, **kwargs) if callable(response_hook): response_hook(span, instance, response) return response @@ -157,6 +223,104 @@ def _traced_execute_pipeline(func, instance, args, kwargs): response_hook(span, instance, response) return response + def _instrumented_redis_call( + service_name, func, name: str, args, **kwargs + ): + generated_id = None + has_execution_index = False + exception = None + status_code = None + should_inject_fault = False + should_abort = True + should_sleep_interval = 0 + vclock = None + origin_vclock = None + execution_index = None + + debug("_instrumented_redis_call entering; method: " + name) + + fuck = context.get_value("suppress_instrumentation") + + # Record that a call is being made to an external service + if not context.get_value("suppress_instrumentation"): + callsite_file, callsite_line, full_traceback_hash = get_full_traceback_hash(service_name) + debug("") + debug("Recording call using Filibuster instrumentation service. ********************") + + # VClock handling + + # TODO: did not do stuff with new-test-excution + + global ei_and_vclock_mutex + ei_and_vclock_mutex.acquire() + + request_id_string = context.get_value(_FILIBUSTER_REQUEST_ID_KEY) + + ## TODO: did not reset local vclock, since we don't reset things for new test executions? + + # Incoming clock from the request that triggered this service to be reached. + incoming_vclock_string = context.get_value(_FILIBUSTER_VCLOCK_KEY) + + # If it's not None, we probably need to merge with our clock, first, since our clock is keeping + # track of *our* requests from this node. + if incoming_vclock_string is not None: + vclocks_by_request = _filibuster_global_context_get_value(_FILIBUSTER_VCLOCK_BY_REQUEST_KEY) + incoming_vclock = vclock_fromstring(incoming_vclock_string) + local_vclock = vclocks_by_request.get(request_id_string, vclock_new()) + new_local_vclock = vclock_merge(incoming_vclock, local_vclock) + vclocks_by_request[request_id_string] = new_local_vclock + _filibuster_global_context_set_value(_FILIBUSTER_VCLOCK_BY_REQUEST_KEY, vclocks_by_request) + + # Finally, advance the clock to account for this request. + vclocks_by_request = _filibuster_global_context_get_value(_FILIBUSTER_VCLOCK_BY_REQUEST_KEY) + local_vclock = vclocks_by_request.get(request_id_string, vclock_new()) + new_local_vclock = vclock_increment(local_vclock, service_name) + vclocks_by_request[request_id_string] = new_local_vclock + _filibuster_global_context_set_value(_FILIBUSTER_VCLOCK_BY_REQUEST_KEY, vclocks_by_request) + + vclock = new_local_vclock + + notice("clock now: " + str(vclocks_by_request.get(request_id_string, vclock_new()))) + + # Maintain the execution index for each request. + + incoming_execution_index_string = context.get_value(_FILIBUSTER_EXECUTION_INDEX_KEY) + + if incoming_execution_index_string is not None: + incoming_execution_index = execution_index_fromstring(incoming_execution_index_string) + else: + execution_indices_by_request = _filibuster_global_context_get_value(_FILIBUSTER_EI_BY_REQUEST_KEY) + incoming_execution_index = execution_indices_by_request.get(request_id_string, + execution_index_new()) + + ## TODO: double check if this url replacement is ok + ## TODO: pretty print execution_index_hash, like in requests code? + execution_index_hash = unique_request_hash( + [full_traceback_hash, 'redis', name, json.dumps(args), json.dumps(kwargs)]) + + execution_indices_by_request = _filibuster_global_context_get_value(_FILIBUSTER_EI_BY_REQUEST_KEY) + execution_indices_by_request[request_id_string] = execution_index_push(execution_index_hash, + incoming_execution_index) + execution_index = execution_indices_by_request[request_id_string] + _filibuster_global_context_set_value(_FILIBUSTER_EI_BY_REQUEST_KEY, execution_indices_by_request) + + ei_and_vclock_mutex.release() + + response = func(*args, **kwargs) + + else: + debug("Instrumentation suppressed, skipping Filibuster instrumentation.") + response = func(*args, **kwargs) + return response + + # For a given request, return a unique hash that can be used to identify it. + def unique_request_hash(args): + for arg in args: + print(arg) + hash_string = "-".join(args) + hex_digest = hashlib.md5(hash_string.encode()).hexdigest() + return hex_digest + pipeline_class = ( "BasePipeline" if redis.VERSION < (3, 0, 0) else "Pipeline" ) @@ -198,6 +362,7 @@ def _instrument(self, **kwargs): tracer, request_hook=kwargs.get("request_hook"), response_hook=kwargs.get("response_hook"), + service_name=kwargs.get("service_name") ) def _uninstrument(self, **kwargs): From e04d90f5527b7a1eb74cd38511adf7f9dfd8920c Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 13 Apr 2022 11:02:07 -0400 Subject: [PATCH 04/10] add apparatus to record call --- filibuster/instrumentation/redis/__init__.py | 147 ++++++++++++++++++- 1 file changed, 142 insertions(+), 5 deletions(-) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index 459e260..2b8a26e 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -83,6 +83,7 @@ def response_hook(span, instance, response): import json import sys +import requests import re import hashlib import os @@ -177,8 +178,13 @@ def _instrument( tracer, request_hook: _RequestHookT = None, response_hook: _ResponseHookT = None, - service_name=None + service_name=None, + filibuster_url=None ): + + def filibuster_create_url(filibuster_url): + return "{}/{}/create".format(filibuster_url, 'filibuster') + def _traced_execute_command(func, instance, args, kwargs): query = _format_command_args(args) name = "" @@ -306,13 +312,143 @@ def _instrumented_redis_call( ei_and_vclock_mutex.release() - response = func(*args, **kwargs) - + # Origin VClock Handling. + + # origin-vclock are used to track the explicit request chain + # that caused this call to be made: more precise than + # happens-before and required for the reduction strategy to + # work. + # + # For example, if Service A does 4 requests, in sequence, + # before making a call to Service B, happens-before can be used + # to show those four requests happened before the call to + # Service B. This is correct: vector/Lamport clock track both + # program order and the communication between nodes in their encoding. + # + # However, for the reduction strategy to work, we need to know + # precisely *what* call in in Service A triggered the call to + # Service B (and, recursively if Service B is to make any + # calls, as well.) This is because the key to the reduction + # strategy is to remove tests from the execution list where + # there is no observable difference at the boundary between the + # two services. Therefore, we need to identify precisely where + # these boundary points are. + # + + # This is a clock that's been received through Flask as part of processing the current request. + # (flask receives context via header and sets into context object; requests reads it.) + incoming_origin_vclock_string = context.get_value(_FILIBUSTER_ORIGIN_VCLOCK_KEY) + debug("** [REQUESTS] [" + service_name + "]: getting incoming origin vclock string: " + str( + incoming_origin_vclock_string)) + + # This isn't used in the record_call, but just propagated through the headers in the subsequent request. + origin_vclock = vclock + + # Record call with the incoming origin clock and advanced clock. + if incoming_origin_vclock_string is not None: + incoming_origin_vclock = vclock_fromstring(incoming_origin_vclock_string) + else: + incoming_origin_vclock = vclock_new() + response = _record_call(service_name, func, name, args, callsite_file, callsite_line, full_traceback_hash, vclock, + incoming_origin_vclock, execution_index_tostring(execution_index), **kwargs) + + if response is not None: + if 'generated_id' in response: + generated_id = response['generated_id'] + + if 'execution_index' in response: + has_execution_index = True + + if 'forced_exception' in response: + exception = response['forced_exception']['name'] + + if 'metadata' in response['forced_exception'] and response['forced_exception'][ + 'metadata'] is not None: + exception_metadata = response['forced_exception']['metadata'] + if 'abort' in exception_metadata and exception_metadata['abort'] is not None: + should_abort = exception_metadata['abort'] + if 'sleep' in exception_metadata and exception_metadata['sleep'] is not None: + should_sleep_interval = exception_metadata['sleep'] + + should_inject_fault = True + + if 'failure_metadata' in response: + if 'return_value' in response['failure_metadata'] and 'status_code' in \ + response['failure_metadata']['return_value']: + status_code = response['failure_metadata']['return_value']['status_code'] + should_inject_fault = True + debug("Finished recording call using Filibuster instrumentation service. ***********") + debug("") else: debug("Instrumentation suppressed, skipping Filibuster instrumentation.") - response = func(*args, **kwargs) + + response = func(*args, **kwargs) + return response + def _record_call(service_name, func, name, args, callsite_file, callsite_line, full_traceback, vclock, origin_vclock, + execution_index, **kwargs): + response = None + parsed_content = None + + try: + debug("Setting Filibuster instrumentation key...") + token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) + + payload = { + 'instrumentation_type': 'invocation', + 'source_service_name': service_name, + 'module': 'redis', + ## TODO: fix this hardcoding + 'method': "execute_command", + 'args': args, + 'kwargs': {}, + 'callsite_file': callsite_file, + 'callsite_line': callsite_line, + 'full_traceback': full_traceback, + 'metadata': {}, + 'vclock': vclock, + 'origin_vclock': origin_vclock, + 'execution_index': execution_index + } + + if 'timeout' in kwargs: + if kwargs['timeout'] is not None: + debug("=> timeout for call is set to " + str(kwargs['timeout'])) + payload['metadata']['timeout'] = kwargs['timeout'] + + if counterexample is not None and counterexample_test_execution is not None: + notice("Using counterexample without contacting server.") + response = should_fail_request_with(payload, counterexample_test_execution.failures) + if response is None: + response = {'execution_index': execution_index} + print(response) + if os.environ.get('DISABLE_SERVER_COMMUNICATION', ''): + warning("Server communication disabled.") + elif counterexample is not None: + notice("Skipping request, replaying from local counterexample.") + else: + requests.post(filibuster_create_url(filibuster_url), json = payload) + except Exception as e: + warning("Exception raised (_record_call)!") + print(e, file=sys.stderr) + return None + finally: + debug("Removing instrumentation key for Filibuster.") + context.detach(token) + + if isinstance(response, dict): + parsed_content = response + elif response is not None: + try: + parsed_content = response.json() + except Exception as e: + warning("Exception raised (_record_call get_json)!") + print(e, file=sys.stderr) + return None + + return parsed_content + # For a given request, return a unique hash that can be used to identify it. def unique_request_hash(args): for arg in args: @@ -362,7 +498,8 @@ def _instrument(self, **kwargs): tracer, request_hook=kwargs.get("request_hook"), response_hook=kwargs.get("response_hook"), - service_name=kwargs.get("service_name") + service_name=kwargs.get("service_name"), + filibuster_url=kwargs.get("filibuster_url") ) def _uninstrument(self, **kwargs): From d4e8518398534182aa8e8146f5acd7dae363bbff Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 13 Apr 2022 14:51:04 -0400 Subject: [PATCH 05/10] successful response case --- filibuster/instrumentation/redis/__init__.py | 105 +++++++++++++++++-- 1 file changed, 99 insertions(+), 6 deletions(-) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index 2b8a26e..8915076 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -81,15 +81,18 @@ def response_hook(span, instance, response): import functools import types import json +import time import sys -import requests import re import hashlib import os from threading import Lock +import requests +from requests.models import Response + from opentelemetry import context from opentelemetry import trace from opentelemetry.instrumentation.instrumentor import BaseInstrumentor @@ -181,6 +184,8 @@ def _instrument( service_name=None, filibuster_url=None ): + def filibuster_update_url(filibuster_url): + return "{}/{}/update".format(filibuster_url, 'filibuster') def filibuster_create_url(filibuster_url): return "{}/{}/create".format(filibuster_url, 'filibuster') @@ -245,8 +250,6 @@ def _instrumented_redis_call( debug("_instrumented_redis_call entering; method: " + name) - fuck = context.get_value("suppress_instrumentation") - # Record that a call is being made to an external service if not context.get_value("suppress_instrumentation"): callsite_file, callsite_line, full_traceback_hash = get_full_traceback_hash(service_name) @@ -382,9 +385,62 @@ def _instrumented_redis_call( else: debug("Instrumentation suppressed, skipping Filibuster instrumentation.") - response = func(*args, **kwargs) - - return response + try: + ## TODO: don't need to do this + # debug("Setting Filibuster instrumentation key...") + # token = context.attach(context.set_value(_FILIBUSTER_SUPPRESS_REQUESTS_INSTRUMENTATION_KEY, True)) + + if not should_inject_fault: + # no need to propagate vclock and origin vclock forward, + # since redis-server won't use it + result = func(*args, **kwargs) + elif should_inject_fault and not should_abort: + # If we should delay the request to simulate timeouts, do it. + if should_sleep_interval != 0: + time.sleep(should_sleep_interval) + + # no need to propagate vclock and origin vclock forward, + # since redis-server won't use it + result = func(*args, **kwargs) + else: + # Return entirely fake response and do not make request. + # + # Since this isn't a real result object, there's some attribute that's + # being set to None and that's causing -- for these requests -- the opentelemetry + # to not be able to report this correctly with the following error in the output: + # + # "Invalid type NoneType for attribute value. + # Expected one of ['bool', 'str', 'int', 'float'] or a sequence of those types" + # + # I'm going to ignore this for now, because if we reorder the instrumentation + # so that the opentelemetry is installed *before* the Filibuster instrumentation + # we should be able to avoid this -- it's because we're returning an invalid + # object through the opentelemetry instrumentation. + # + result = Response() + except Exception as exc: + exception = exc + result = getattr(exc, "response", None) + finally: + debug("Removing instrumentation key for Filibuster.") + # context.detach(token) + + # Result was an actual response. + if isinstance(result, Response) and (exception is None or exception == "None"): + debug("_instrumented_requests_call got response!") + + if has_execution_index: + _update_execution_index() + + # Notify the filibuster server of the actual response. + if generated_id is not None: + _record_successful_response(generated_id, execution_index_tostring(execution_index), vclock, + result) + + # Result was an exception + ## TODO + + return result def _record_call(service_name, func, name, args, callsite_file, callsite_line, full_traceback, vclock, origin_vclock, execution_index, **kwargs): @@ -449,6 +505,43 @@ def _record_call(service_name, func, name, args, callsite_file, callsite_line, f return parsed_content + def _update_execution_index(): + global ei_and_vclock_mutex + + ei_and_vclock_mutex.acquire() + + execution_indices_by_request = _filibuster_global_context_get_value(_FILIBUSTER_EI_BY_REQUEST_KEY) + request_id_string = context.get_value(_FILIBUSTER_REQUEST_ID_KEY) + if request_id_string in execution_indices_by_request: + execution_indices_by_request[request_id_string] = execution_index_pop( + execution_indices_by_request[request_id_string]) + _filibuster_global_context_set_value(_FILIBUSTER_EI_BY_REQUEST_KEY, execution_indices_by_request) + + ei_and_vclock_mutex.release() + + def _record_successful_response(generated_id, execution_index, vclock, result): + # assumes no asynchrony or threads at calling service. + + if not (os.environ.get('DISABLE_SERVER_COMMUNICATION', '')) and counterexample is None: + try: + debug("Setting Filibuster instrumentation key...") + token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) + + return_value = { + ## TODO + } + payload = { + ## TODO + } + requests.post(filibuster_update_url(filibuster_url), json=payload) + except Exception as e: + warning("Exception raised (_record_successful_response)!") + print(e, file=sys.stderr) + finally: + debug("Removing instrumentation key for Filibuster.") + context.detach(token) + + return True # For a given request, return a unique hash that can be used to identify it. def unique_request_hash(args): for arg in args: From a7b8a04bed84c766e61df3c39b6c9aac1f1ff832 Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 13 Apr 2022 15:34:44 -0400 Subject: [PATCH 06/10] added code to record successful response --- filibuster/instrumentation/redis/__init__.py | 56 +++++++++++++++++--- 1 file changed, 49 insertions(+), 7 deletions(-) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index 8915076..bee0449 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -417,7 +417,7 @@ def _instrumented_redis_call( # we should be able to avoid this -- it's because we're returning an invalid # object through the opentelemetry instrumentation. # - result = Response() + result = None except Exception as exc: exception = exc result = getattr(exc, "response", None) @@ -426,7 +426,7 @@ def _instrumented_redis_call( # context.detach(token) # Result was an actual response. - if isinstance(result, Response) and (exception is None or exception == "None"): + if not result is None and (exception is None or exception == "None"): debug("_instrumented_requests_call got response!") if has_execution_index: @@ -437,8 +437,47 @@ def _instrumented_redis_call( _record_successful_response(generated_id, execution_index_tostring(execution_index), vclock, result) - # Result was an exception + # Result was an exception. ## TODO + # if exception is not None and exception != "None": + # if isinstance(exception, str): + # exception_class = eval(exception) + # exception = exception_class() + # use_traceback = False + # else: + # if context.get_value(_FILIBUSTER_INSTRUMENTATION_KEY): + # # If the Filibuster instrumentation call failed, ignore. This just means + # # that the test server is unavailable. + # warning("Filibuster instrumentation server unreachable, ignoring...") + # warning("If fault injection is enabled... this indicates that something isn't working properly.") + # else: + # try: + # exception_info = exception.rsplit('.', 1) + # m = importlib.import_module(exception_info[0]) + # exception = getattr(m, exception_info[1]) + # except Exception: + # warning("Couldn't get actual exception due to exception parse error.") + + # use_traceback = True + + # if not context.get_value(_FILIBUSTER_INSTRUMENTATION_KEY): + # debug("got exception!") + # debug("=> exception: " + str(exception)) + + # if has_execution_index: + # _update_execution_index(self) + + # # Notify the filibuster server of the actual exception we encountered. + # if generated_id is not None: + # _record_exceptional_response(self, generated_id, execution_index_tostring(execution_index), vclock, + # exception, should_sleep_interval, should_abort) + + # if use_traceback: + # raise exception.with_traceback(exception.__traceback__) + # else: + # raise exception + + # debug("_instrumented_requests_call exiting; method: " + method + " url: " + url) return result @@ -527,11 +566,14 @@ def _record_successful_response(generated_id, execution_index, vclock, result): debug("Setting Filibuster instrumentation key...") token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) - return_value = { - ## TODO - } + ## TODO: double check if this is ok + return_value = { result } payload = { - ## TODO + 'instrumentation_type': 'invocation_complete', + 'generated_id': generated_id, + 'execution_index': execution_index, + 'vclock': vclock, + 'return_value': return_value } requests.post(filibuster_update_url(filibuster_url), json=payload) except Exception as e: From 27f32e83b0a41dcc9af4d09c685775a7293e9b7e Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Tue, 19 Apr 2022 18:22:10 -0400 Subject: [PATCH 07/10] added exception stuff to redis --- filibuster/instrumentation/redis/__init__.py | 190 ++++++++++++------- 1 file changed, 120 insertions(+), 70 deletions(-) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index bee0449..3429273 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -105,6 +105,8 @@ def response_hook(span, instance, response): from opentelemetry.semconv.trace import SpanAttributes from opentelemetry.trace import Span +import importlib + from filibuster.global_context import get_value as _filibuster_global_context_get_value from filibuster.global_context import set_value as _filibuster_global_context_set_value from filibuster.execution_index import execution_index_new, execution_index_fromstring, \ @@ -118,10 +120,6 @@ def response_hook(span, instance, response): from filibuster.datatypes import TestExecution from filibuster.instrumentation.helpers import get_full_traceback_hash -# A key to a context variable to avoid creating duplicate spans when instrumenting -# both, Session.request and Session.send, since Session.request calls into Session.send -_FILIBUSTER_SUPPRESS_REQUESTS_INSTRUMENTATION_KEY = "filibuster_suppress_requests_instrumentation" - # We do not want to instrument the instrumentation, so use this key to detect when we # are inside of a Filibuster instrumentation call to suppress further instrumentation. _FILIBUSTER_INSTRUMENTATION_KEY = "filibuster_instrumentation" @@ -190,6 +188,9 @@ def filibuster_update_url(filibuster_url): def filibuster_create_url(filibuster_url): return "{}/{}/create".format(filibuster_url, 'filibuster') + def filibuster_new_test_execution_url(filibuster_url, service_name): + return "{}/{}/new-test-execution/{}".format(filibuster_url, 'filibuster', service_name) + def _traced_execute_command(func, instance, args, kwargs): query = _format_command_args(args) name = "" @@ -240,7 +241,6 @@ def _instrumented_redis_call( generated_id = None has_execution_index = False exception = None - status_code = None should_inject_fault = False should_abort = True should_sleep_interval = 0 @@ -258,14 +258,36 @@ def _instrumented_redis_call( # VClock handling - # TODO: did not do stuff with new-test-excution + # Figure out if we should reset the node's vector clock, which should happen in between test executions. + debug("Setting Filibuster instrumentation key...") + token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) + + response = None + if not (os.environ.get('DISABLE_SERVER_COMMUNICATION', '')) and counterexample is None: + requests.post('get', filibuster_new_test_execution_url(filibuster_url, service_name)) + if response is not None: + response = response.json() + + debug("Removing instrumentation key for Filibuster.") + context.detach(token) + reset_local_vclock = False + if response and ('new-test-execution' in response) and (response['new-test-execution']): + reset_local_vclock = True global ei_and_vclock_mutex ei_and_vclock_mutex.acquire() request_id_string = context.get_value(_FILIBUSTER_REQUEST_ID_KEY) - ## TODO: did not reset local vclock, since we don't reset things for new test executions? + if reset_local_vclock: + # Reset everything, since there is a new test execution. + debug("New test execution. Resetting vclocks_by_request and execution_indices_by_request.") + + vclocks_by_request = {request_id_string: vclock_new()} + _filibuster_global_context_set_value(_FILIBUSTER_VCLOCK_BY_REQUEST_KEY, vclocks_by_request) + + execution_indices_by_request = {request_id_string: execution_index_new()} + _filibuster_global_context_set_value(_FILIBUSTER_EI_BY_REQUEST_KEY, execution_indices_by_request) # Incoming clock from the request that triggered this service to be reached. incoming_vclock_string = context.get_value(_FILIBUSTER_VCLOCK_KEY) @@ -304,8 +326,10 @@ def _instrumented_redis_call( ## TODO: double check if this url replacement is ok ## TODO: pretty print execution_index_hash, like in requests code? + ## TODO: need use url instead (on two different redis nodes) + ## TODO: name could be execute_command instead execution_index_hash = unique_request_hash( - [full_traceback_hash, 'redis', name, json.dumps(args), json.dumps(kwargs)]) + [full_traceback_hash, 'redis', 'execute_command', json.dumps(args), json.dumps(kwargs)]) execution_indices_by_request = _filibuster_global_context_get_value(_FILIBUSTER_EI_BY_REQUEST_KEY) execution_indices_by_request[request_id_string] = execution_index_push(execution_index_hash, @@ -375,21 +399,18 @@ def _instrumented_redis_call( should_inject_fault = True + ## TODO: fix once Filibuster injects faulty responses if 'failure_metadata' in response: if 'return_value' in response['failure_metadata'] and 'status_code' in \ response['failure_metadata']['return_value']: - status_code = response['failure_metadata']['return_value']['status_code'] should_inject_fault = True + debug("Finished recording call using Filibuster instrumentation service. ***********") debug("") else: debug("Instrumentation suppressed, skipping Filibuster instrumentation.") try: - ## TODO: don't need to do this - # debug("Setting Filibuster instrumentation key...") - # token = context.attach(context.set_value(_FILIBUSTER_SUPPRESS_REQUESTS_INSTRUMENTATION_KEY, True)) - if not should_inject_fault: # no need to propagate vclock and origin vclock forward, # since redis-server won't use it @@ -404,19 +425,7 @@ def _instrumented_redis_call( result = func(*args, **kwargs) else: # Return entirely fake response and do not make request. - # - # Since this isn't a real result object, there's some attribute that's - # being set to None and that's causing -- for these requests -- the opentelemetry - # to not be able to report this correctly with the following error in the output: - # - # "Invalid type NoneType for attribute value. - # Expected one of ['bool', 'str', 'int', 'float'] or a sequence of those types" - # - # I'm going to ignore this for now, because if we reorder the instrumentation - # so that the opentelemetry is installed *before* the Filibuster instrumentation - # we should be able to avoid this -- it's because we're returning an invalid - # object through the opentelemetry instrumentation. - # + ## TODO: fix (redis may return 0 for s.ismember if it doesn't find it) result = None except Exception as exc: exception = exc @@ -438,46 +447,45 @@ def _instrumented_redis_call( result) # Result was an exception. - ## TODO - # if exception is not None and exception != "None": - # if isinstance(exception, str): - # exception_class = eval(exception) - # exception = exception_class() - # use_traceback = False - # else: - # if context.get_value(_FILIBUSTER_INSTRUMENTATION_KEY): - # # If the Filibuster instrumentation call failed, ignore. This just means - # # that the test server is unavailable. - # warning("Filibuster instrumentation server unreachable, ignoring...") - # warning("If fault injection is enabled... this indicates that something isn't working properly.") - # else: - # try: - # exception_info = exception.rsplit('.', 1) - # m = importlib.import_module(exception_info[0]) - # exception = getattr(m, exception_info[1]) - # except Exception: - # warning("Couldn't get actual exception due to exception parse error.") - - # use_traceback = True - - # if not context.get_value(_FILIBUSTER_INSTRUMENTATION_KEY): - # debug("got exception!") - # debug("=> exception: " + str(exception)) - - # if has_execution_index: - # _update_execution_index(self) - - # # Notify the filibuster server of the actual exception we encountered. - # if generated_id is not None: - # _record_exceptional_response(self, generated_id, execution_index_tostring(execution_index), vclock, - # exception, should_sleep_interval, should_abort) - - # if use_traceback: - # raise exception.with_traceback(exception.__traceback__) - # else: - # raise exception - - # debug("_instrumented_requests_call exiting; method: " + method + " url: " + url) + if exception is not None and exception != "None": + if isinstance(exception, str): + exception_class = eval(exception) + exception = exception_class() + use_traceback = False + else: + if context.get_value(_FILIBUSTER_INSTRUMENTATION_KEY): + # If the Filibuster instrumentation call failed, ignore. This just means + # that the test server is unavailable. + warning("Filibuster instrumentation server unreachable, ignoring...") + warning("If fault injection is enabled... this indicates that something isn't working properly.") + else: + try: + exception_info = exception.rsplit('.', 1) + m = importlib.import_module(exception_info[0]) + exception = getattr(m, exception_info[1]) + except Exception: + warning("Couldn't get actual exception due to exception parse error.") + + use_traceback = True + + if not context.get_value(_FILIBUSTER_INSTRUMENTATION_KEY): + debug("got exception!") + debug("=> exception: " + str(exception)) + + if has_execution_index: + _update_execution_index() + + # Notify the filibuster server of the actual exception we encountered. + if generated_id is not None: + _record_exceptional_response(generated_id, execution_index_tostring(execution_index), vclock, + exception, should_sleep_interval, should_abort) + + if use_traceback: + raise exception.with_traceback(exception.__traceback__) + else: + raise exception + + debug("_instrumented_requests_call exiting; method: " + name) return result @@ -517,7 +525,7 @@ def _record_call(service_name, func, name, args, callsite_file, callsite_line, f response = should_fail_request_with(payload, counterexample_test_execution.failures) if response is None: response = {'execution_index': execution_index} - print(response) + if os.environ.get('DISABLE_SERVER_COMMUNICATION', ''): warning("Server communication disabled.") elif counterexample is not None: @@ -567,7 +575,11 @@ def _record_successful_response(generated_id, execution_index, vclock, result): token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) ## TODO: double check if this is ok - return_value = { result } + return_value = { + '__class__': str(result.__class__.__name__), + 'value': result, + 'text': hashlib.md5(result.text.encode()).hexdigest() + } payload = { 'instrumentation_type': 'invocation_complete', 'generated_id': generated_id, @@ -584,10 +596,48 @@ def _record_successful_response(generated_id, execution_index, vclock, result): context.detach(token) return True + + def _record_exceptional_response(generated_id, execution_index, vclock, exception, should_sleep_interval, + should_abort): + # assumes no asynchrony or threads at calling service. + if not (os.environ.get('DISABLE_SERVER_COMMUNICATION', '')): + try: + debug("Setting Filibuster instrumentation key...") + token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) + + exception_to_string = str(type(exception)) + parsed_exception_string = re.findall(r"'(.*?)'", exception_to_string, re.DOTALL)[0] + payload = { + 'instrumentation_type': 'invocation_complete', + 'generated_id': generated_id, + 'execution_index': execution_index, + 'vclock': vclock, + 'exception': { + 'name': parsed_exception_string, + 'metadata': { + + } + } + } + + if should_sleep_interval > 0: + payload['exception']['metadata']['sleep'] = should_sleep_interval + + if should_abort is not True: + payload['exception']['metadata']['abort'] = should_abort + + requests.post(filibuster_update_url(filibuster_url), json=payload) + except Exception as e: + warning("Exception raised (_record_exceptional_response)!") + print(e, file=sys.stderr) + finally: + debug("Removing instrumentation key for Filibuster.") + context.detach(token) + + return True + # For a given request, return a unique hash that can be used to identify it. def unique_request_hash(args): - for arg in args: - print(arg) hash_string = "-".join(args) hex_digest = hashlib.md5(hash_string.encode()).hexdigest() return hex_digest From 9cced1cebd239240937c19779d4cac3e8f9b9c6e Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 20 Apr 2022 14:25:28 -0400 Subject: [PATCH 08/10] added certain redis exceptions to possible faults --- default-analysis.json | 31 +++++++++++++++++ filibuster/analysis/__init__.py | 36 +++++++++++++++++++- filibuster/instrumentation/redis/__init__.py | 19 ++++++----- 3 files changed, 76 insertions(+), 10 deletions(-) diff --git a/default-analysis.json b/default-analysis.json index 5f4bae5..0353a5f 100644 --- a/default-analysis.json +++ b/default-analysis.json @@ -58,6 +58,20 @@ } ] }, + "python.redis": { + "pattern": "redis\\.execute\\_command", + "exceptions": [ + { + "name": "redis.exceptions.ConnectionError" + }, + { + "name": "redis.exceptions.TimeoutError" + }, + { + "name": "redis.exceptions.ResponseError" + } + ] + }, "http": { "pattern": "(((requests\\.(get|put|post|head))|(WebClient\\.(GET|PUT|POST|HEAD))))", "errors": [ @@ -99,5 +113,22 @@ ] } ] + }, + "redis": { + "pattern": "redis\\.execute\\_command", + "errors": [ + { + "service_name": ".*", + "types": [ + { + "exception": { + "metadata": { + "code": "redis.exceptions.RedisError" + } + } + } + ] + } + ] } } \ No newline at end of file diff --git a/filibuster/analysis/__init__.py b/filibuster/analysis/__init__.py index 5a2ade3..af09c4e 100644 --- a/filibuster/analysis/__init__.py +++ b/filibuster/analysis/__init__.py @@ -160,6 +160,25 @@ def java_header_to_status_code(constant): print("Found constant with no match: " + constant) raise Exception("Analysis failed: unknown java header constant.") +def add_python_redis_exceptions(): + # Setup. + instrumentation['python.redis'] = {} + instrumentation['python.redis']['pattern'] = "redis\\.execute\\_command" + instrumentation['python.redis']['exceptions'] = [] + + # Base exceptions. + # ConnectionError raised when the redis-server has disconnected + instrumentation['python.redis']['exceptions'].append( + {'name': 'redis.exceptions.ConnectionError'}) + instrumentation['python.redis']['exceptions'].append( + {'name': 'redis.exceptions.TimeoutError'}) + # ResponseError raised when set operations are used on a key whose value is not a set + instrumentation['python.redis']['exceptions'].append( + {'name': 'redis.exceptions.ResponseError'}) + # TODO: check this + # base class of all redis exceptions, don't raise this by default + # instrumentation['python.redis']['exceptions'].append( + # {'name': 'redis.exceptions.RedisError'}) def exception_to_status_code(exception): if exception == "Forbidden": @@ -248,6 +267,14 @@ def analyze_python(service, filename): info("* identified HTTP error: " + str(general_failure_type)) services_and_errors['types'].append(general_failure_type) + ## TODO: not exactly sure what default failure type I should put here + redis_failure_type = {'exception': {'metadata': {'code': 'redis.exceptions.RedisError'}}} + for services_and_errors in instrumentation['redis']['errors']: + if services_and_errors['service_name'] == service: + if failure_type not in services_and_errors['types']: + info("* identified redis error: " + str(redis_failure_type)) + services_and_errors['types'].append(redis_failure_type) + # Get pb2 specific errors. file = open(filename, "r") for line in file: @@ -330,6 +357,8 @@ def analyze_services_directory(output, directory): # Add Java grpc callsite exceptions. add_java_grpc_exceptions() + add_python_redis_exceptions() + ################################################################################################################## # Fill out placeholder information for parsable file. ################################################################################################################## @@ -340,7 +369,7 @@ def analyze_services_directory(output, directory): if 'grpc' not in instrumentation: instrumentation['grpc'] = {} - instrumentation['grpc']['pattern'] = "((grpc\\.insecure\_channel)|(.*Service/.*))" + instrumentation['grpc']['pattern'] = "((grpc\\.insecure\\_channel)|(.*Service/.*))" if 'errors' not in instrumentation['http']: instrumentation['http']['errors'] = [] @@ -348,6 +377,10 @@ def analyze_services_directory(output, directory): if 'errors' not in instrumentation['grpc']: instrumentation['grpc']['errors'] = [] + if 'redis' not in instrumentation: + instrumentation['redis'] = {} + instrumentation['redis']['pattern'] = "redis\\.execute\\_command" + ################################################################################################################## # Analyze each service. ################################################################################################################## @@ -370,6 +403,7 @@ def analyze_services_directory(output, directory): for service in services: instrumentation['http']['errors'].append({'service_name': service, 'types': []}) instrumentation['grpc']['errors'].append({'service_name': service, 'types': []}) + instrumentation['redis']['errors'].append({'service_name': service, 'types': []}) for service in services: service_directory = os.path.join(directory, service) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index 3429273..bc04824 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -76,6 +76,7 @@ def response_hook(span, instance, response): from typing import Any, Collection import redis +import redis.exceptions from wrapt import wrap_function_wrapper import functools @@ -210,7 +211,7 @@ def _traced_execute_command(func, instance, args, kwargs): if callable(request_hook): request_hook(span, instance, args, kwargs) - response = _instrumented_redis_call(service_name, func, name, args, **kwargs) + response = _instrumented_redis_call(service_name, func, "execute_command", args, **kwargs) if callable(response_hook): response_hook(span, instance, response) return response @@ -236,7 +237,7 @@ def _traced_execute_pipeline(func, instance, args, kwargs): return response def _instrumented_redis_call( - service_name, func, name: str, args, **kwargs + service_name, func, method: str, args, **kwargs ): generated_id = None has_execution_index = False @@ -248,7 +249,7 @@ def _instrumented_redis_call( origin_vclock = None execution_index = None - debug("_instrumented_redis_call entering; method: " + name) + debug("_instrumented_redis_call entering; method: " + method) # Record that a call is being made to an external service if not context.get_value("suppress_instrumentation"): @@ -264,7 +265,7 @@ def _instrumented_redis_call( response = None if not (os.environ.get('DISABLE_SERVER_COMMUNICATION', '')) and counterexample is None: - requests.post('get', filibuster_new_test_execution_url(filibuster_url, service_name)) + requests.get(filibuster_new_test_execution_url(filibuster_url, service_name)) if response is not None: response = response.json() @@ -329,7 +330,7 @@ def _instrumented_redis_call( ## TODO: need use url instead (on two different redis nodes) ## TODO: name could be execute_command instead execution_index_hash = unique_request_hash( - [full_traceback_hash, 'redis', 'execute_command', json.dumps(args), json.dumps(kwargs)]) + [full_traceback_hash, 'redis', method, json.dumps(args), json.dumps(kwargs)]) execution_indices_by_request = _filibuster_global_context_get_value(_FILIBUSTER_EI_BY_REQUEST_KEY) execution_indices_by_request[request_id_string] = execution_index_push(execution_index_hash, @@ -376,7 +377,7 @@ def _instrumented_redis_call( incoming_origin_vclock = vclock_fromstring(incoming_origin_vclock_string) else: incoming_origin_vclock = vclock_new() - response = _record_call(service_name, func, name, args, callsite_file, callsite_line, full_traceback_hash, vclock, + response = _record_call(service_name, func, method, args, callsite_file, callsite_line, full_traceback_hash, vclock, incoming_origin_vclock, execution_index_tostring(execution_index), **kwargs) if response is not None: @@ -485,11 +486,11 @@ def _instrumented_redis_call( else: raise exception - debug("_instrumented_requests_call exiting; method: " + name) + debug("_instrumented_requests_call exiting; method: " + method) return result - def _record_call(service_name, func, name, args, callsite_file, callsite_line, full_traceback, vclock, origin_vclock, + def _record_call(service_name, func, method, args, callsite_file, callsite_line, full_traceback, vclock, origin_vclock, execution_index, **kwargs): response = None parsed_content = None @@ -531,7 +532,7 @@ def _record_call(service_name, func, name, args, callsite_file, callsite_line, f elif counterexample is not None: notice("Skipping request, replaying from local counterexample.") else: - requests.post(filibuster_create_url(filibuster_url), json = payload) + requests.put(filibuster_create_url(filibuster_url), json = payload) except Exception as e: warning("Exception raised (_record_call)!") print(e, file=sys.stderr) From 1945ffb1399368b905cfed6c7b43efd89f34cde7 Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 27 Apr 2022 09:14:02 -0400 Subject: [PATCH 09/10] fixed the exception code, should be working now --- filibuster/analysis/__init__.py | 17 ----------------- filibuster/instrumentation/redis/__init__.py | 18 +++++++++++++----- 2 files changed, 13 insertions(+), 22 deletions(-) diff --git a/filibuster/analysis/__init__.py b/filibuster/analysis/__init__.py index af09c4e..778b12f 100644 --- a/filibuster/analysis/__init__.py +++ b/filibuster/analysis/__init__.py @@ -175,10 +175,6 @@ def add_python_redis_exceptions(): # ResponseError raised when set operations are used on a key whose value is not a set instrumentation['python.redis']['exceptions'].append( {'name': 'redis.exceptions.ResponseError'}) - # TODO: check this - # base class of all redis exceptions, don't raise this by default - # instrumentation['python.redis']['exceptions'].append( - # {'name': 'redis.exceptions.RedisError'}) def exception_to_status_code(exception): if exception == "Forbidden": @@ -267,14 +263,6 @@ def analyze_python(service, filename): info("* identified HTTP error: " + str(general_failure_type)) services_and_errors['types'].append(general_failure_type) - ## TODO: not exactly sure what default failure type I should put here - redis_failure_type = {'exception': {'metadata': {'code': 'redis.exceptions.RedisError'}}} - for services_and_errors in instrumentation['redis']['errors']: - if services_and_errors['service_name'] == service: - if failure_type not in services_and_errors['types']: - info("* identified redis error: " + str(redis_failure_type)) - services_and_errors['types'].append(redis_failure_type) - # Get pb2 specific errors. file = open(filename, "r") for line in file: @@ -377,10 +365,6 @@ def analyze_services_directory(output, directory): if 'errors' not in instrumentation['grpc']: instrumentation['grpc']['errors'] = [] - if 'redis' not in instrumentation: - instrumentation['redis'] = {} - instrumentation['redis']['pattern'] = "redis\\.execute\\_command" - ################################################################################################################## # Analyze each service. ################################################################################################################## @@ -403,7 +387,6 @@ def analyze_services_directory(output, directory): for service in services: instrumentation['http']['errors'].append({'service_name': service, 'types': []}) instrumentation['grpc']['errors'].append({'service_name': service, 'types': []}) - instrumentation['redis']['errors'].append({'service_name': service, 'types': []}) for service in services: service_directory = os.path.join(directory, service) diff --git a/filibuster/instrumentation/redis/__init__.py b/filibuster/instrumentation/redis/__init__.py index bc04824..ccd1de3 100644 --- a/filibuster/instrumentation/redis/__init__.py +++ b/filibuster/instrumentation/redis/__init__.py @@ -265,7 +265,7 @@ def _instrumented_redis_call( response = None if not (os.environ.get('DISABLE_SERVER_COMMUNICATION', '')) and counterexample is None: - requests.get(filibuster_new_test_execution_url(filibuster_url, service_name)) + response = requests.get(filibuster_new_test_execution_url(filibuster_url, service_name)) if response is not None: response = response.json() @@ -433,7 +433,6 @@ def _instrumented_redis_call( result = getattr(exc, "response", None) finally: debug("Removing instrumentation key for Filibuster.") - # context.detach(token) # Result was an actual response. if not result is None and (exception is None or exception == "None"): @@ -532,7 +531,7 @@ def _record_call(service_name, func, method, args, callsite_file, callsite_line, elif counterexample is not None: notice("Skipping request, replaying from local counterexample.") else: - requests.put(filibuster_create_url(filibuster_url), json = payload) + response = requests.put(filibuster_create_url(filibuster_url), json = payload) except Exception as e: warning("Exception raised (_record_call)!") print(e, file=sys.stderr) @@ -576,11 +575,20 @@ def _record_successful_response(generated_id, execution_index, vclock, result): token = context.attach(context.set_value(_FILIBUSTER_INSTRUMENTATION_KEY, True)) ## TODO: double check if this is ok + # Need to change the 'text' field depending on the type of result + class_name = str(result.__class__.__name__) + if class_name == "dict": + text = json.dumps(result).encode() + elif class_name == "list": + text =','.join([str(elem) for elem in result]).encode() + else: + text = str(result).encode() return_value = { - '__class__': str(result.__class__.__name__), + '__class__': class_name, 'value': result, - 'text': hashlib.md5(result.text.encode()).hexdigest() + 'text': hashlib.md5(text).hexdigest() } + payload = { 'instrumentation_type': 'invocation_complete', 'generated_id': generated_id, From 4c4211c8d0644f84448caf9e6d24ff5a5456d50b Mon Sep 17 00:00:00 2001 From: Eunice Chen Date: Wed, 27 Apr 2022 11:20:28 -0400 Subject: [PATCH 10/10] forgot to add default-analysis.json --- default-analysis.json | 17 ----------------- 1 file changed, 17 deletions(-) diff --git a/default-analysis.json b/default-analysis.json index 0353a5f..64133f7 100644 --- a/default-analysis.json +++ b/default-analysis.json @@ -113,22 +113,5 @@ ] } ] - }, - "redis": { - "pattern": "redis\\.execute\\_command", - "errors": [ - { - "service_name": ".*", - "types": [ - { - "exception": { - "metadata": { - "code": "redis.exceptions.RedisError" - } - } - } - ] - } - ] } } \ No newline at end of file