diff options
author | David Garcia Quintas <dgq@google.com> | 2016-03-31 09:43:03 -0700 |
---|---|---|
committer | David Garcia Quintas <dgq@google.com> | 2016-03-31 09:43:03 -0700 |
commit | 2877d7398e9f6bf815af3f42f21cb4c3d917be2e (patch) | |
tree | 7a08a3e544cf8583d4bb2405bb53e48ebdac5a65 /src/core/lib/census | |
parent | 2c73b22885ea07ecd338e71a0baf5cdcfc45605a (diff) | |
parent | 2b9012362ca03181eefaec22238c6100e77bb736 (diff) |
Merge branch 'master' of github.com:grpc/grpc into lb_resolvers
Diffstat (limited to 'src/core/lib/census')
-rw-r--r-- | src/core/lib/census/README.md | 76 | ||||
-rw-r--r-- | src/core/lib/census/aggregation.h | 66 | ||||
-rw-r--r-- | src/core/lib/census/context.c | 509 | ||||
-rw-r--r-- | src/core/lib/census/grpc_context.c | 53 | ||||
-rw-r--r-- | src/core/lib/census/grpc_filter.c | 198 | ||||
-rw-r--r-- | src/core/lib/census/grpc_filter.h | 44 | ||||
-rw-r--r-- | src/core/lib/census/grpc_plugin.c | 70 | ||||
-rw-r--r-- | src/core/lib/census/grpc_plugin.h | 40 | ||||
-rw-r--r-- | src/core/lib/census/initialize.c | 54 | ||||
-rw-r--r-- | src/core/lib/census/mlog.c | 600 | ||||
-rw-r--r-- | src/core/lib/census/mlog.h | 95 | ||||
-rw-r--r-- | src/core/lib/census/operation.c | 63 | ||||
-rw-r--r-- | src/core/lib/census/placeholders.c | 109 | ||||
-rw-r--r-- | src/core/lib/census/rpc_metric_id.h | 51 | ||||
-rw-r--r-- | src/core/lib/census/tracing.c | 45 |
15 files changed, 2073 insertions, 0 deletions
diff --git a/src/core/lib/census/README.md b/src/core/lib/census/README.md new file mode 100644 index 0000000000..fb615a2194 --- /dev/null +++ b/src/core/lib/census/README.md @@ -0,0 +1,76 @@ +<!--- + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +--> + +# Census - a resource measurement and tracing system + +This directory contains code for Census, which will ultimately provide the +following features for any gRPC-using system: +* A [dapper](http://research.google.com/pubs/pub36356.html)-like tracing + system, enabling tracing across a distributed infrastructure. +* RPC statistics and measurements for key metrics, such as latency, bytes + transferred, number of errors etc. +* Resource measurement framework which can be used for measuring custom + metrics. Through the use of [tags](#Tags), these can be broken down across + the entire distributed stack. +* Easy integration of the above with + [Google Cloud Trace](https://cloud.google.com/tools/cloud-trace) and + [Google Cloud Monitoring](https://cloud.google.com/monitoring/). + +## Concepts + +### Context + +### Operations + +### Tags + +### Metrics + +## API + +### Internal/RPC API + +### External/Client API + +### RPC API + +## Files in this directory + +Note that files and functions in this directory can be split into two +categories: +* Files that define core census library functions. Functions etc. in these + files are named census\_\*, and constitute the core census library + functionality. At some time in the future, these will become a standalone + library. +* Files that define functions etc. that provide a convenient interface between + grpc and the core census functionality. These files are all named + grpc\_\*.{c,h}, and define function names beginning with grpc\_census\_\*. + diff --git a/src/core/lib/census/aggregation.h b/src/core/lib/census/aggregation.h new file mode 100644 index 0000000000..f353368b97 --- /dev/null +++ b/src/core/lib/census/aggregation.h @@ -0,0 +1,66 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <stddef.h> + +#ifndef GRPC_CORE_LIB_CENSUS_AGGREGATION_H +#define GRPC_CORE_LIB_CENSUS_AGGREGATION_H + +/** Structure used to describe an aggregation type. */ +struct census_aggregation_ops { + /* Create a new aggregation. The pointer returned can be used in future calls + to clone(), free(), record(), data() and reset(). */ + void *(*create)(const void *create_arg); + /* Make a copy of an aggregation created by create() */ + void *(*clone)(const void *aggregation); + /* Destroy an aggregation created by create() */ + void (*free)(void *aggregation); + /* Record a new value against aggregation. */ + void (*record)(void *aggregation, double value); + /* Return current aggregation data. The caller must cast this object into + the correct type for the aggregation result. The object returned can be + freed by using free_data(). */ + void *(*data)(const void *aggregation); + /* free data returned by data() */ + void (*free_data)(void *data); + /* Reset an aggregation to default (zero) values. */ + void (*reset)(void *aggregation); + /* Merge 'from' aggregation into 'to'. Both aggregations must be compatible */ + void (*merge)(void *to, const void *from); + /* Fill buffer with printable string version of aggregation contents. For + debugging only. Returns the number of bytes added to buffer (a value == n + implies the buffer was of insufficient size). */ + size_t (*print)(const void *aggregation, char *buffer, size_t n); +}; + +#endif /* GRPC_CORE_LIB_CENSUS_AGGREGATION_H */ diff --git a/src/core/lib/census/context.c b/src/core/lib/census/context.c new file mode 100644 index 0000000000..5a118f46a9 --- /dev/null +++ b/src/core/lib/census/context.c @@ -0,0 +1,509 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/census.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/port_platform.h> +#include <grpc/support/useful.h> +#include <stdbool.h> +#include <string.h> +#include "src/core/lib/support/string.h" + +// Functions in this file support the public context API, including +// encoding/decoding as part of context propagation across RPC's. The overall +// requirements (in approximate priority order) for the +// context representation: +// 1. Efficient conversion to/from wire format +// 2. Minimal bytes used on-wire +// 3. Efficient context creation +// 4. Efficient lookup of tag value for a key +// 5. Efficient iteration over tags +// 6. Minimal memory footprint +// +// Notes on tradeoffs/decisions: +// * tag includes 1 byte length of key, as well as nil-terminating byte. These +// are to aid in efficient parsing and the ability to directly return key +// strings. This is more important than saving a single byte/tag on the wire. +// * The wire encoding uses only single byte values. This eliminates the need +// to handle endian-ness conversions. It also means there is a hard upper +// limit of 255 for both CENSUS_MAX_TAG_KV_LEN and CENSUS_MAX_PROPAGATED_TAGS. +// * Keep all tag information (keys/values/flags) in a single memory buffer, +// that can be directly copied to the wire. + +// min and max valid chars in tag keys and values. All printable ASCII is OK. +#define MIN_VALID_TAG_CHAR 32 // ' ' +#define MAX_VALID_TAG_CHAR 126 // '~' + +// Structure representing a set of tags. Essentially a count of number of tags +// present, and pointer to a chunk of memory that contains the per-tag details. +struct tag_set { + int ntags; // number of tags. + int ntags_alloc; // ntags + number of deleted tags (total number of tags + // in all of kvm). This will always be == ntags, except during the process + // of building a new tag set. + size_t kvm_size; // number of bytes allocated for key/value storage. + size_t kvm_used; // number of bytes of used key/value memory + char *kvm; // key/value memory. Consists of repeated entries of: + // Offset Size Description + // 0 1 Key length, including trailing 0. (K) + // 1 1 Value length, including trailing 0 (V) + // 2 1 Flags + // 3 K Key bytes + // 3 + K V Value bytes + // + // We refer to the first 3 entries as the 'tag header'. If extra values are + // introduced in the header, you will need to modify the TAG_HEADER_SIZE + // constant, the raw_tag structure (and everything that uses it) and the + // encode/decode functions appropriately. +}; + +// Number of bytes in tag header. +#define TAG_HEADER_SIZE 3 // key length (1) + value length (1) + flags (1) +// Offsets to tag header entries. +#define KEY_LEN_OFFSET 0 +#define VALUE_LEN_OFFSET 1 +#define FLAG_OFFSET 2 + +// raw_tag represents the raw-storage form of a tag in the kvm of a tag_set. +struct raw_tag { + uint8_t key_len; + uint8_t value_len; + uint8_t flags; + char *key; + char *value; +}; + +// Use a reserved flag bit for indication of deleted tag. +#define CENSUS_TAG_DELETED CENSUS_TAG_RESERVED +#define CENSUS_TAG_IS_DELETED(flags) (flags & CENSUS_TAG_DELETED) + +// Primary representation of a context. Composed of 2 underlying tag_set +// structs, one each for propagated and local (non-propagated) tags. This is +// to efficiently support tag encoding/decoding. +// TODO(aveitch): need to add tracing id's/structure. +struct census_context { + struct tag_set tags[2]; + census_context_status status; +}; + +// Indices into the tags member of census_context +#define PROPAGATED_TAGS 0 +#define LOCAL_TAGS 1 + +// Validate (check all characters are in range and size is less than limit) a +// key or value string. Returns 0 if the string is invalid, or the length +// (including terminator) if valid. +static size_t validate_tag(const char *kv) { + size_t len = 1; + char ch; + while ((ch = *kv++) != 0) { + if (ch < MIN_VALID_TAG_CHAR || ch > MAX_VALID_TAG_CHAR) { + return 0; + } + len++; + } + if (len > CENSUS_MAX_TAG_KV_LEN) { + return 0; + } + return len; +} + +// Extract a raw tag given a pointer (raw) to the tag header. Allow for some +// extra bytes in the tag header (see encode/decode functions for usage: this +// allows for future expansion of the tag header). +static char *decode_tag(struct raw_tag *tag, char *header, int offset) { + tag->key_len = (uint8_t)(*header++); + tag->value_len = (uint8_t)(*header++); + tag->flags = (uint8_t)(*header++); + header += offset; + tag->key = header; + header += tag->key_len; + tag->value = header; + return header + tag->value_len; +} + +// Make a copy (in 'to') of an existing tag_set. +static void tag_set_copy(struct tag_set *to, const struct tag_set *from) { + memcpy(to, from, sizeof(struct tag_set)); + to->kvm = gpr_malloc(to->kvm_size); + memcpy(to->kvm, from->kvm, from->kvm_used); +} + +// Delete a tag from a tag_set, if it exists (returns true if it did). +static bool tag_set_delete_tag(struct tag_set *tags, const char *key, + size_t key_len) { + char *kvp = tags->kvm; + for (int i = 0; i < tags->ntags_alloc; i++) { + uint8_t *flags = (uint8_t *)(kvp + FLAG_OFFSET); + struct raw_tag tag; + kvp = decode_tag(&tag, kvp, 0); + if (CENSUS_TAG_IS_DELETED(tag.flags)) continue; + if ((key_len == tag.key_len) && (memcmp(key, tag.key, key_len) == 0)) { + *flags |= CENSUS_TAG_DELETED; + tags->ntags--; + return true; + } + } + return false; +} + +// Delete a tag from a context, return true if it existed. +static bool context_delete_tag(census_context *context, const census_tag *tag, + size_t key_len) { + return ( + tag_set_delete_tag(&context->tags[LOCAL_TAGS], tag->key, key_len) || + tag_set_delete_tag(&context->tags[PROPAGATED_TAGS], tag->key, key_len)); +} + +// Add a tag to a tag_set. Return true on success, false if the tag could +// not be added because of constraints on tag set size. This function should +// not be called if the tag may already exist (in a non-deleted state) in +// the tag_set, as that would result in two tags with the same key. +static bool tag_set_add_tag(struct tag_set *tags, const census_tag *tag, + size_t key_len, size_t value_len) { + if (tags->ntags == CENSUS_MAX_PROPAGATED_TAGS) { + return false; + } + const size_t tag_size = key_len + value_len + TAG_HEADER_SIZE; + if (tags->kvm_used + tag_size > tags->kvm_size) { + // allocate new memory if needed + tags->kvm_size += 2 * CENSUS_MAX_TAG_KV_LEN + TAG_HEADER_SIZE; + char *new_kvm = gpr_malloc(tags->kvm_size); + memcpy(new_kvm, tags->kvm, tags->kvm_used); + gpr_free(tags->kvm); + tags->kvm = new_kvm; + } + char *kvp = tags->kvm + tags->kvm_used; + *kvp++ = (char)key_len; + *kvp++ = (char)value_len; + // ensure reserved flags are not used. + *kvp++ = (char)(tag->flags & (CENSUS_TAG_PROPAGATE | CENSUS_TAG_STATS)); + memcpy(kvp, tag->key, key_len); + kvp += key_len; + memcpy(kvp, tag->value, value_len); + tags->kvm_used += tag_size; + tags->ntags++; + tags->ntags_alloc++; + return true; +} + +// Add/modify/delete a tag to/in a context. Caller must validate that tag key +// etc. are valid. +static void context_modify_tag(census_context *context, const census_tag *tag, + size_t key_len, size_t value_len) { + // First delete the tag if it is already present. + bool deleted = context_delete_tag(context, tag, key_len); + bool added = false; + if (CENSUS_TAG_IS_PROPAGATED(tag->flags)) { + added = tag_set_add_tag(&context->tags[PROPAGATED_TAGS], tag, key_len, + value_len); + } else { + added = + tag_set_add_tag(&context->tags[LOCAL_TAGS], tag, key_len, value_len); + } + + if (deleted) { + context->status.n_modified_tags++; + } else { + if (added) { + context->status.n_added_tags++; + } else { + context->status.n_ignored_tags++; + } + } +} + +// Remove memory used for deleted tags from a tag set. Basic algorithm: +// 1) Walk through tag set to find first deleted tag. Record where it is. +// 2) Find the next not-deleted tag. Copy all of kvm from there to the end +// "over" the deleted tags +// 3) repeat #1 and #2 until we have seen all tags +// 4) if we are still looking for a not-deleted tag, then all the end portion +// of the kvm is deleted. Just reduce the used amount of memory by the +// appropriate amount. +static void tag_set_flatten(struct tag_set *tags) { + if (tags->ntags == tags->ntags_alloc) return; + bool found_deleted = false; // found a deleted tag. + char *kvp = tags->kvm; + char *dbase = NULL; // record location of deleted tag + for (int i = 0; i < tags->ntags_alloc; i++) { + struct raw_tag tag; + char *next_kvp = decode_tag(&tag, kvp, 0); + if (found_deleted) { + if (!CENSUS_TAG_IS_DELETED(tag.flags)) { + ptrdiff_t reduce = kvp - dbase; // #bytes in deleted tags + GPR_ASSERT(reduce > 0); + ptrdiff_t copy_size = tags->kvm + tags->kvm_used - kvp; + GPR_ASSERT(copy_size > 0); + memmove(dbase, kvp, (size_t)copy_size); + tags->kvm_used -= (size_t)reduce; + next_kvp -= reduce; + found_deleted = false; + } + } else { + if (CENSUS_TAG_IS_DELETED(tag.flags)) { + dbase = kvp; + found_deleted = true; + } + } + kvp = next_kvp; + } + if (found_deleted) { + GPR_ASSERT(dbase > tags->kvm); + tags->kvm_used = (size_t)(dbase - tags->kvm); + } + tags->ntags_alloc = tags->ntags; +} + +census_context *census_context_create(const census_context *base, + const census_tag *tags, int ntags, + census_context_status const **status) { + census_context *context = gpr_malloc(sizeof(census_context)); + // If we are given a base, copy it into our new tag set. Otherwise set it + // to zero/NULL everything. + if (base == NULL) { + memset(context, 0, sizeof(census_context)); + } else { + tag_set_copy(&context->tags[PROPAGATED_TAGS], &base->tags[PROPAGATED_TAGS]); + tag_set_copy(&context->tags[LOCAL_TAGS], &base->tags[LOCAL_TAGS]); + memset(&context->status, 0, sizeof(context->status)); + } + // Walk over the additional tags and, for those that aren't invalid, modify + // the context to add/replace/delete as required. + for (int i = 0; i < ntags; i++) { + const census_tag *tag = &tags[i]; + size_t key_len = validate_tag(tag->key); + // ignore the tag if it is invalid or too short. + if (key_len <= 1) { + context->status.n_invalid_tags++; + } else { + if (tag->value != NULL) { + size_t value_len = validate_tag(tag->value); + if (value_len != 0) { + context_modify_tag(context, tag, key_len, value_len); + } else { + context->status.n_invalid_tags++; + } + } else { + if (context_delete_tag(context, tag, key_len)) { + context->status.n_deleted_tags++; + } + } + } + } + // Remove any deleted tags, update status if needed, and return. + tag_set_flatten(&context->tags[PROPAGATED_TAGS]); + tag_set_flatten(&context->tags[LOCAL_TAGS]); + context->status.n_propagated_tags = context->tags[PROPAGATED_TAGS].ntags; + context->status.n_local_tags = context->tags[LOCAL_TAGS].ntags; + if (status) { + *status = &context->status; + } + return context; +} + +const census_context_status *census_context_get_status( + const census_context *context) { + return &context->status; +} + +void census_context_destroy(census_context *context) { + gpr_free(context->tags[PROPAGATED_TAGS].kvm); + gpr_free(context->tags[LOCAL_TAGS].kvm); + gpr_free(context); +} + +void census_context_initialize_iterator(const census_context *context, + census_context_iterator *iterator) { + iterator->context = context; + iterator->index = 0; + if (context->tags[PROPAGATED_TAGS].ntags != 0) { + iterator->base = PROPAGATED_TAGS; + iterator->kvm = context->tags[PROPAGATED_TAGS].kvm; + } else if (context->tags[LOCAL_TAGS].ntags != 0) { + iterator->base = LOCAL_TAGS; + iterator->kvm = context->tags[LOCAL_TAGS].kvm; + } else { + iterator->base = -1; + } +} + +int census_context_next_tag(census_context_iterator *iterator, + census_tag *tag) { + if (iterator->base < 0) { + return 0; + } + struct raw_tag raw; + iterator->kvm = decode_tag(&raw, iterator->kvm, 0); + tag->key = raw.key; + tag->value = raw.value; + tag->flags = raw.flags; + if (++iterator->index == iterator->context->tags[iterator->base].ntags) { + do { + if (iterator->base == LOCAL_TAGS) { + iterator->base = -1; + return 1; + } + } while (iterator->context->tags[++iterator->base].ntags == 0); + iterator->index = 0; + iterator->kvm = iterator->context->tags[iterator->base].kvm; + } + return 1; +} + +// Find a tag in a tag_set by key. Return true if found, false otherwise. +static bool tag_set_get_tag(const struct tag_set *tags, const char *key, + size_t key_len, census_tag *tag) { + char *kvp = tags->kvm; + for (int i = 0; i < tags->ntags; i++) { + struct raw_tag raw; + kvp = decode_tag(&raw, kvp, 0); + if (key_len == raw.key_len && memcmp(raw.key, key, key_len) == 0) { + tag->key = raw.key; + tag->value = raw.value; + tag->flags = raw.flags; + return true; + } + } + return false; +} + +int census_context_get_tag(const census_context *context, const char *key, + census_tag *tag) { + size_t key_len = strlen(key) + 1; + if (key_len == 1) { + return 0; + } + if (tag_set_get_tag(&context->tags[PROPAGATED_TAGS], key, key_len, tag) || + tag_set_get_tag(&context->tags[LOCAL_TAGS], key, key_len, tag)) { + return 1; + } + return 0; +} + +// Context encoding and decoding functions. +// +// Wire format for tag_set's on the wire: +// +// First, a tag set header: +// +// offset bytes description +// 0 1 version number +// 1 1 number of bytes in this header. This allows for future +// expansion. +// 2 1 number of bytes in each tag header. +// 3 1 ntags value from tag set. +// +// This is followed by the key/value memory from struct tag_set. + +#define ENCODED_VERSION 0 // Version number +#define ENCODED_HEADER_SIZE 4 // size of tag set header + +// Encode a tag set. Returns 0 if buffer is too small. +static size_t tag_set_encode(const struct tag_set *tags, char *buffer, + size_t buf_size) { + if (buf_size < ENCODED_HEADER_SIZE + tags->kvm_used) { + return 0; + } + buf_size -= ENCODED_HEADER_SIZE; + *buffer++ = (char)ENCODED_VERSION; + *buffer++ = (char)ENCODED_HEADER_SIZE; + *buffer++ = (char)TAG_HEADER_SIZE; + *buffer++ = (char)tags->ntags; + if (tags->ntags == 0) { + return ENCODED_HEADER_SIZE; + } + memcpy(buffer, tags->kvm, tags->kvm_used); + return ENCODED_HEADER_SIZE + tags->kvm_used; +} + +size_t census_context_encode(const census_context *context, char *buffer, + size_t buf_size) { + return tag_set_encode(&context->tags[PROPAGATED_TAGS], buffer, buf_size); +} + +// Decode a tag set. +static void tag_set_decode(struct tag_set *tags, const char *buffer, + size_t size) { + uint8_t version = (uint8_t)(*buffer++); + uint8_t header_size = (uint8_t)(*buffer++); + uint8_t tag_header_size = (uint8_t)(*buffer++); + tags->ntags = tags->ntags_alloc = (int)(*buffer++); + if (tags->ntags == 0) { + tags->ntags_alloc = 0; + tags->kvm_size = 0; + tags->kvm_used = 0; + tags->kvm = NULL; + return; + } + if (header_size != ENCODED_HEADER_SIZE) { + GPR_ASSERT(version != ENCODED_VERSION); + GPR_ASSERT(ENCODED_HEADER_SIZE < header_size); + buffer += (header_size - ENCODED_HEADER_SIZE); + } + tags->kvm_used = size - header_size; + tags->kvm_size = tags->kvm_used + CENSUS_MAX_TAG_KV_LEN; + tags->kvm = gpr_malloc(tags->kvm_size); + if (tag_header_size != TAG_HEADER_SIZE) { + // something new in the tag information. I don't understand it, so + // don't copy it over. + GPR_ASSERT(version != ENCODED_VERSION); + GPR_ASSERT(tag_header_size > TAG_HEADER_SIZE); + char *kvp = tags->kvm; + for (int i = 0; i < tags->ntags; i++) { + memcpy(kvp, buffer, TAG_HEADER_SIZE); + kvp += header_size; + struct raw_tag raw; + buffer = + decode_tag(&raw, (char *)buffer, tag_header_size - TAG_HEADER_SIZE); + memcpy(kvp, raw.key, (size_t)raw.key_len + raw.value_len); + kvp += raw.key_len + raw.value_len; + } + } else { + memcpy(tags->kvm, buffer, tags->kvm_used); + } +} + +census_context *census_context_decode(const char *buffer, size_t size) { + census_context *context = gpr_malloc(sizeof(census_context)); + memset(&context->tags[LOCAL_TAGS], 0, sizeof(struct tag_set)); + if (buffer == NULL) { + memset(&context->tags[PROPAGATED_TAGS], 0, sizeof(struct tag_set)); + } else { + tag_set_decode(&context->tags[PROPAGATED_TAGS], buffer, size); + } + memset(&context->status, 0, sizeof(context->status)); + context->status.n_propagated_tags = context->tags[PROPAGATED_TAGS].ntags; + return context; +} diff --git a/src/core/lib/census/grpc_context.c b/src/core/lib/census/grpc_context.c new file mode 100644 index 0000000000..457c176355 --- /dev/null +++ b/src/core/lib/census/grpc_context.c @@ -0,0 +1,53 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/census.h> +#include <grpc/grpc.h> +#include "src/core/lib/surface/api_trace.h" +#include "src/core/lib/surface/call.h" + +void grpc_census_call_set_context(grpc_call *call, census_context *context) { + GRPC_API_TRACE("grpc_census_call_set_context(call=%p, census_context=%p)", 2, + (call, context)); + if (census_enabled() == CENSUS_FEATURE_NONE) { + return; + } + if (context != NULL) { + grpc_call_context_set(call, GRPC_CONTEXT_TRACING, context, NULL); + } +} + +census_context *grpc_census_call_get_context(grpc_call *call) { + GRPC_API_TRACE("grpc_census_call_get_context(call=%p)", 1, (call)); + return (census_context *)grpc_call_context_get(call, GRPC_CONTEXT_TRACING); +} diff --git a/src/core/lib/census/grpc_filter.c b/src/core/lib/census/grpc_filter.c new file mode 100644 index 0000000000..d27d789aa1 --- /dev/null +++ b/src/core/lib/census/grpc_filter.c @@ -0,0 +1,198 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/census/grpc_filter.h" + +#include <stdio.h> +#include <string.h> + +#include <grpc/census.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/slice.h> +#include <grpc/support/time.h> + +#include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/statistics/census_interface.h" +#include "src/core/lib/statistics/census_rpc_stats.h" +#include "src/core/lib/transport/static_metadata.h" + +typedef struct call_data { + census_op_id op_id; + census_context *ctxt; + gpr_timespec start_ts; + int error; + + /* recv callback */ + grpc_metadata_batch *recv_initial_metadata; + grpc_closure *on_done_recv; + grpc_closure finish_recv; +} call_data; + +typedef struct channel_data { uint8_t unused; } channel_data; + +static void extract_and_annotate_method_tag(grpc_metadata_batch *md, + call_data *calld, + channel_data *chand) { + grpc_linked_mdelem *m; + for (m = md->list.head; m != NULL; m = m->next) { + if (m->md->key == GRPC_MDSTR_PATH) { + gpr_log(GPR_DEBUG, "%s", + (const char *)GPR_SLICE_START_PTR(m->md->value->slice)); + /* Add method tag here */ + } + } +} + +static void client_mutate_op(grpc_call_element *elem, + grpc_transport_stream_op *op) { + call_data *calld = elem->call_data; + channel_data *chand = elem->channel_data; + if (op->send_initial_metadata) { + extract_and_annotate_method_tag(op->send_initial_metadata, calld, chand); + } +} + +static void client_start_transport_op(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_transport_stream_op *op) { + client_mutate_op(elem, op); + grpc_call_next_op(exec_ctx, elem, op); +} + +static void server_on_done_recv(grpc_exec_ctx *exec_ctx, void *ptr, + bool success) { + grpc_call_element *elem = ptr; + call_data *calld = elem->call_data; + channel_data *chand = elem->channel_data; + if (success) { + extract_and_annotate_method_tag(calld->recv_initial_metadata, calld, chand); + } + calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, success); +} + +static void server_mutate_op(grpc_call_element *elem, + grpc_transport_stream_op *op) { + call_data *calld = elem->call_data; + if (op->recv_initial_metadata) { + /* substitute our callback for the op callback */ + calld->recv_initial_metadata = op->recv_initial_metadata; + calld->on_done_recv = op->recv_initial_metadata_ready; + op->recv_initial_metadata_ready = &calld->finish_recv; + } +} + +static void server_start_transport_op(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_transport_stream_op *op) { + /* TODO(ctiller): this code fails. I don't know why. I expect it's + incomplete, and someone should look at it soon. + + call_data *calld = elem->call_data; + GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); */ + server_mutate_op(elem, op); + grpc_call_next_op(exec_ctx, elem, op); +} + +static void client_init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { + call_data *d = elem->call_data; + GPR_ASSERT(d != NULL); + memset(d, 0, sizeof(*d)); + d->start_ts = gpr_now(GPR_CLOCK_REALTIME); +} + +static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem) { + call_data *d = elem->call_data; + GPR_ASSERT(d != NULL); + /* TODO(hongyu): record rpc client stats and census_rpc_end_op here */ +} + +static void server_init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { + call_data *d = elem->call_data; + GPR_ASSERT(d != NULL); + memset(d, 0, sizeof(*d)); + d->start_ts = gpr_now(GPR_CLOCK_REALTIME); + /* TODO(hongyu): call census_tracing_start_op here. */ + grpc_closure_init(&d->finish_recv, server_on_done_recv, elem); +} + +static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem) { + call_data *d = elem->call_data; + GPR_ASSERT(d != NULL); + /* TODO(hongyu): record rpc server stats and census_tracing_end_op here */ +} + +static void init_channel_elem(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem, + grpc_channel_element_args *args) { + channel_data *chand = elem->channel_data; + GPR_ASSERT(chand != NULL); +} + +static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem) { + channel_data *chand = elem->channel_data; + GPR_ASSERT(chand != NULL); +} + +const grpc_channel_filter grpc_client_census_filter = { + client_start_transport_op, + grpc_channel_next_op, + sizeof(call_data), + client_init_call_elem, + grpc_call_stack_ignore_set_pollset, + client_destroy_call_elem, + sizeof(channel_data), + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "census-client"}; + +const grpc_channel_filter grpc_server_census_filter = { + server_start_transport_op, + grpc_channel_next_op, + sizeof(call_data), + server_init_call_elem, + grpc_call_stack_ignore_set_pollset, + server_destroy_call_elem, + sizeof(channel_data), + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "census-server"}; diff --git a/src/core/lib/census/grpc_filter.h b/src/core/lib/census/grpc_filter.h new file mode 100644 index 0000000000..7ceafe56e4 --- /dev/null +++ b/src/core/lib/census/grpc_filter.h @@ -0,0 +1,44 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_CENSUS_GRPC_FILTER_H +#define GRPC_CORE_LIB_CENSUS_GRPC_FILTER_H + +#include "src/core/lib/channel/channel_stack.h" + +/* Census filters: provides tracing and stats collection functionalities. It + needs to reside right below the surface filter in the channel stack. */ +extern const grpc_channel_filter grpc_client_census_filter; +extern const grpc_channel_filter grpc_server_census_filter; + +#endif /* GRPC_CORE_LIB_CENSUS_GRPC_FILTER_H */ diff --git a/src/core/lib/census/grpc_plugin.c b/src/core/lib/census/grpc_plugin.c new file mode 100644 index 0000000000..12aca76745 --- /dev/null +++ b/src/core/lib/census/grpc_plugin.c @@ -0,0 +1,70 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/census/grpc_plugin.h" + +#include <limits.h> + +#include <grpc/census.h> + +#include "src/core/lib/census/grpc_filter.h" +#include "src/core/lib/channel/channel_stack_builder.h" +#include "src/core/lib/surface/channel_init.h" + +static bool maybe_add_census_filter(grpc_channel_stack_builder *builder, + void *arg_must_be_null) { + const grpc_channel_args *args = + grpc_channel_stack_builder_get_channel_arguments(builder); + if (grpc_channel_args_is_census_enabled(args)) { + return grpc_channel_stack_builder_prepend_filter( + builder, &grpc_client_census_filter, NULL, NULL); + } + return true; +} + +void census_grpc_plugin_init(void) { + /* Only initialize census if no one else has and some features are + * available. */ + if (census_enabled() == CENSUS_FEATURE_NONE && + census_supported() != CENSUS_FEATURE_NONE) { + if (census_initialize(census_supported())) { /* enable all features. */ + gpr_log(GPR_ERROR, "Could not initialize census."); + } + } + grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX, + maybe_add_census_filter, NULL); + grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX, + maybe_add_census_filter, NULL); +} + +void census_grpc_plugin_destroy(void) { census_shutdown(); } diff --git a/src/core/lib/census/grpc_plugin.h b/src/core/lib/census/grpc_plugin.h new file mode 100644 index 0000000000..33e5f0b701 --- /dev/null +++ b/src/core/lib/census/grpc_plugin.h @@ -0,0 +1,40 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_CENSUS_GRPC_PLUGIN_H +#define GRPC_CORE_LIB_CENSUS_GRPC_PLUGIN_H + +void census_grpc_plugin_init(void); +void census_grpc_plugin_destroy(void); + +#endif /* GRPC_CORE_LIB_CENSUS_GRPC_PLUGIN_H */ diff --git a/src/core/lib/census/initialize.c b/src/core/lib/census/initialize.c new file mode 100644 index 0000000000..ce7ec09b89 --- /dev/null +++ b/src/core/lib/census/initialize.c @@ -0,0 +1,54 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/census.h> + +static int features_enabled = CENSUS_FEATURE_NONE; + +int census_initialize(int features) { + if (features_enabled != CENSUS_FEATURE_NONE) { + // Must have been a previous call to census_initialize; return error + return 1; + } + features_enabled = features; + return 0; +} + +void census_shutdown(void) { features_enabled = CENSUS_FEATURE_NONE; } + +int census_supported(void) { + /* TODO(aveitch): improve this as we implement features... */ + return CENSUS_FEATURE_NONE; +} + +int census_enabled(void) { return features_enabled; } diff --git a/src/core/lib/census/mlog.c b/src/core/lib/census/mlog.c new file mode 100644 index 0000000000..9d47e80297 --- /dev/null +++ b/src/core/lib/census/mlog.c @@ -0,0 +1,600 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +// Implements an efficient in-memory log, optimized for multiple writers and +// a single reader. Available log space is divided up in blocks of +// CENSUS_LOG_2_MAX_RECORD_SIZE bytes. A block can be in one of the following +// three data structures: +// - Free blocks (free_block_list) +// - Blocks with unread data (dirty_block_list) +// - Blocks currently attached to cores (core_local_blocks[]) +// +// census_log_start_write() moves a block from core_local_blocks[] to the end of +// dirty_block_list when block: +// - is out-of-space OR +// - has an incomplete record (an incomplete record occurs when a thread calls +// census_log_start_write() and is context-switched before calling +// census_log_end_write() +// So, blocks in dirty_block_list are ordered, from oldest to newest, by the +// time when block is detached from the core. +// +// census_log_read_next() first iterates over dirty_block_list and then +// core_local_blocks[]. It moves completely read blocks from dirty_block_list +// to free_block_list. Blocks in core_local_blocks[] are not freed, even when +// completely read. +// +// If the log is configured to discard old records and free_block_list is empty, +// census_log_start_write() iterates over dirty_block_list to allocate a +// new block. It moves the oldest available block (no pending read/write) to +// core_local_blocks[]. +// +// core_local_block_struct is used to implement a map from core id to the block +// associated with that core. This mapping is advisory. It is possible that the +// block returned by this mapping is no longer associated with that core. This +// mapping is updated, lazily, by census_log_start_write(). +// +// Locking in block struct: +// +// Exclusive g_log.lock must be held before calling any functions operating on +// block structs except census_log_start_write() and census_log_end_write(). +// +// Writes to a block are serialized via writer_lock. census_log_start_write() +// acquires this lock and census_log_end_write() releases it. On failure to +// acquire the lock, writer allocates a new block for the current core and +// updates core_local_block accordingly. +// +// Simultaneous read and write access is allowed. Readers can safely read up to +// committed bytes (bytes_committed). +// +// reader_lock protects the block, currently being read, from getting recycled. +// start_read() acquires reader_lock and end_read() releases the lock. +// +// Read/write access to a block is disabled via try_disable_access(). It returns +// with both writer_lock and reader_lock held. These locks are subsequently +// released by enable_access() to enable access to the block. +// +// A note on naming: Most function/struct names are prepended by cl_ +// (shorthand for census_log). Further, functions that manipulate structures +// include the name of the structure, which will be passed as the first +// argument. E.g. cl_block_initialize() will initialize a cl_block. + +#include "src/core/lib/census/mlog.h" +#include <grpc/support/alloc.h> +#include <grpc/support/atm.h> +#include <grpc/support/cpu.h> +#include <grpc/support/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/useful.h> +#include <stdbool.h> +#include <string.h> + +// End of platform specific code + +typedef struct census_log_block_list_struct { + struct census_log_block_list_struct* next; + struct census_log_block_list_struct* prev; + struct census_log_block* block; +} cl_block_list_struct; + +typedef struct census_log_block { + // Pointer to underlying buffer. + char* buffer; + gpr_atm writer_lock; + gpr_atm reader_lock; + // Keeps completely written bytes. Declared atomic because accessed + // simultaneously by reader and writer. + gpr_atm bytes_committed; + // Bytes already read. + size_t bytes_read; + // Links for list. + cl_block_list_struct link; +// We want this structure to be cacheline aligned. We assume the following +// sizes for the various parts on 32/64bit systems: +// type 32b size 64b size +// char* 4 8 +// 3x gpr_atm 12 24 +// size_t 4 8 +// cl_block_list_struct 12 24 +// TOTAL 32 64 +// +// Depending on the size of our cacheline and the architecture, we +// selectively add char buffering to this structure. The size is checked +// via assert in census_log_initialize(). +#if defined(GPR_ARCH_64) +#define CL_BLOCK_PAD_SIZE (GPR_CACHELINE_SIZE - 64) +#else +#if defined(GPR_ARCH_32) +#define CL_BLOCK_PAD_SIZE (GPR_CACHELINE_SIZE - 32) +#else +#error "Unknown architecture" +#endif +#endif +#if CL_BLOCK_PAD_SIZE > 0 + char padding[CL_BLOCK_PAD_SIZE]; +#endif +} cl_block; + +// A list of cl_blocks, doubly-linked through cl_block::link. +typedef struct census_log_block_list { + int32_t count; // Number of items in list. + cl_block_list_struct ht; // head/tail of linked list. +} cl_block_list; + +// Cacheline aligned block pointers to avoid false sharing. Block pointer must +// be initialized via set_block(), before calling other functions +typedef struct census_log_core_local_block { + gpr_atm block; +// Ensure cachline alignment: we assume sizeof(gpr_atm) == 4 or 8 +#if defined(GPR_ARCH_64) +#define CL_CORE_LOCAL_BLOCK_PAD_SIZE (GPR_CACHELINE_SIZE - 8) +#else +#if defined(GPR_ARCH_32) +#define CL_CORE_LOCAL_BLOCK_PAD_SIZE (GPR_CACHELINE_SIZE - 4) +#else +#error "Unknown architecture" +#endif +#endif +#if CL_CORE_LOCAL_BLOCK_PAD_SIZE > 0 + char padding[CL_CORE_LOCAL_BLOCK_PAD_SIZE]; +#endif +} cl_core_local_block; + +struct census_log { + int discard_old_records; + // Number of cores (aka hardware-contexts) + unsigned num_cores; + // number of CENSUS_LOG_2_MAX_RECORD_SIZE blocks in log + uint32_t num_blocks; + cl_block* blocks; // Block metadata. + cl_core_local_block* core_local_blocks; // Keeps core to block mappings. + gpr_mu lock; + int initialized; // has log been initialized? + // Keeps the state of the reader iterator. A value of 0 indicates that + // iterator has reached the end. census_log_init_reader() resets the value + // to num_core to restart iteration. + uint32_t read_iterator_state; + // Points to the block being read. If non-NULL, the block is locked for + // reading(block_being_read_->reader_lock is held). + cl_block* block_being_read; + char* buffer; + cl_block_list free_block_list; + cl_block_list dirty_block_list; + gpr_atm out_of_space_count; +}; + +// Single internal log. +static struct census_log g_log; + +// Functions that operate on an atomic memory location used as a lock. + +// Returns non-zero if lock is acquired. +static int cl_try_lock(gpr_atm* lock) { return gpr_atm_acq_cas(lock, 0, 1); } + +static void cl_unlock(gpr_atm* lock) { gpr_atm_rel_store(lock, 0); } + +// Functions that operate on cl_core_local_block's. + +static void cl_core_local_block_set_block(cl_core_local_block* clb, + cl_block* block) { + gpr_atm_rel_store(&clb->block, (gpr_atm)block); +} + +static cl_block* cl_core_local_block_get_block(cl_core_local_block* clb) { + return (cl_block*)gpr_atm_acq_load(&clb->block); +} + +// Functions that operate on cl_block_list_struct's. + +static void cl_block_list_struct_initialize(cl_block_list_struct* bls, + cl_block* block) { + bls->next = bls->prev = bls; + bls->block = block; +} + +// Functions that operate on cl_block_list's. + +static void cl_block_list_initialize(cl_block_list* list) { + list->count = 0; + cl_block_list_struct_initialize(&list->ht, NULL); +} + +// Returns head of *this, or NULL if empty. +static cl_block* cl_block_list_head(cl_block_list* list) { + return list->ht.next->block; +} + +// Insert element *e after *pos. +static void cl_block_list_insert(cl_block_list* list, cl_block_list_struct* pos, + cl_block_list_struct* e) { + list->count++; + e->next = pos->next; + e->prev = pos; + e->next->prev = e; + e->prev->next = e; +} + +// Insert block at the head of the list +static void cl_block_list_insert_at_head(cl_block_list* list, cl_block* block) { + cl_block_list_insert(list, &list->ht, &block->link); +} + +// Insert block at the tail of the list. +static void cl_block_list_insert_at_tail(cl_block_list* list, cl_block* block) { + cl_block_list_insert(list, list->ht.prev, &block->link); +} + +// Removes block *b. Requires *b be in the list. +static void cl_block_list_remove(cl_block_list* list, cl_block* b) { + list->count--; + b->link.next->prev = b->link.prev; + b->link.prev->next = b->link.next; +} + +// Functions that operate on cl_block's + +static void cl_block_initialize(cl_block* block, char* buffer) { + block->buffer = buffer; + gpr_atm_rel_store(&block->writer_lock, 0); + gpr_atm_rel_store(&block->reader_lock, 0); + gpr_atm_rel_store(&block->bytes_committed, 0); + block->bytes_read = 0; + cl_block_list_struct_initialize(&block->link, block); +} + +// Guards against exposing partially written buffer to the reader. +static void cl_block_set_bytes_committed(cl_block* block, + size_t bytes_committed) { + gpr_atm_rel_store(&block->bytes_committed, (gpr_atm)bytes_committed); +} + +static size_t cl_block_get_bytes_committed(cl_block* block) { + return (size_t)gpr_atm_acq_load(&block->bytes_committed); +} + +// Tries to disable future read/write access to this block. Succeeds if: +// - no in-progress write AND +// - no in-progress read AND +// - 'discard_data' set to true OR no unread data +// On success, clears the block state and returns with writer_lock_ and +// reader_lock_ held. These locks are released by a subsequent +// cl_block_access_enable() call. +static bool cl_block_try_disable_access(cl_block* block, int discard_data) { + if (!cl_try_lock(&block->writer_lock)) { + return false; + } + if (!cl_try_lock(&block->reader_lock)) { + cl_unlock(&block->writer_lock); + return false; + } + if (!discard_data && + (block->bytes_read != cl_block_get_bytes_committed(block))) { + cl_unlock(&block->reader_lock); + cl_unlock(&block->writer_lock); + return false; + } + cl_block_set_bytes_committed(block, 0); + block->bytes_read = 0; + return true; +} + +static void cl_block_enable_access(cl_block* block) { + cl_unlock(&block->reader_lock); + cl_unlock(&block->writer_lock); +} + +// Returns with writer_lock held. +static void* cl_block_start_write(cl_block* block, size_t size) { + if (!cl_try_lock(&block->writer_lock)) { + return NULL; + } + size_t bytes_committed = cl_block_get_bytes_committed(block); + if (bytes_committed + size > CENSUS_LOG_MAX_RECORD_SIZE) { + cl_unlock(&block->writer_lock); + return NULL; + } + return block->buffer + bytes_committed; +} + +// Releases writer_lock and increments committed bytes by 'bytes_written'. +// 'bytes_written' must be <= 'size' specified in the corresponding +// StartWrite() call. This function is thread-safe. +static void cl_block_end_write(cl_block* block, size_t bytes_written) { + cl_block_set_bytes_committed( + block, cl_block_get_bytes_committed(block) + bytes_written); + cl_unlock(&block->writer_lock); +} + +// Returns a pointer to the first unread byte in buffer. The number of bytes +// available are returned in 'bytes_available'. Acquires reader lock that is +// released by a subsequent cl_block_end_read() call. Returns NULL if: +// - read in progress +// - no data available +static void* cl_block_start_read(cl_block* block, size_t* bytes_available) { + if (!cl_try_lock(&block->reader_lock)) { + return NULL; + } + // bytes_committed may change from under us. Use bytes_available to update + // bytes_read below. + size_t bytes_committed = cl_block_get_bytes_committed(block); + GPR_ASSERT(bytes_committed >= block->bytes_read); + *bytes_available = bytes_committed - block->bytes_read; + if (*bytes_available == 0) { + cl_unlock(&block->reader_lock); + return NULL; + } + void* record = block->buffer + block->bytes_read; + block->bytes_read += *bytes_available; + return record; +} + +static void cl_block_end_read(cl_block* block) { + cl_unlock(&block->reader_lock); +} + +// Internal functions operating on g_log + +// Allocates a new free block (or recycles an available dirty block if log is +// configured to discard old records). Returns NULL if out-of-space. +static cl_block* cl_allocate_block(void) { + cl_block* block = cl_block_list_head(&g_log.free_block_list); + if (block != NULL) { + cl_block_list_remove(&g_log.free_block_list, block); + return block; + } + if (!g_log.discard_old_records) { + // No free block and log is configured to keep old records. + return NULL; + } + // Recycle dirty block. Start from the oldest. + for (block = cl_block_list_head(&g_log.dirty_block_list); block != NULL; + block = block->link.next->block) { + if (cl_block_try_disable_access(block, 1 /* discard data */)) { + cl_block_list_remove(&g_log.dirty_block_list, block); + return block; + } + } + return NULL; +} + +// Allocates a new block and updates core id => block mapping. 'old_block' +// points to the block that the caller thinks is attached to +// 'core_id'. 'old_block' may be NULL. Returns true if: +// - allocated a new block OR +// - 'core_id' => 'old_block' mapping changed (another thread allocated a +// block before lock was acquired). +static bool cl_allocate_core_local_block(uint32_t core_id, + cl_block* old_block) { + // Now that we have the lock, check if core-local mapping has changed. + cl_core_local_block* core_local_block = &g_log.core_local_blocks[core_id]; + cl_block* block = cl_core_local_block_get_block(core_local_block); + if ((block != NULL) && (block != old_block)) { + return true; + } + if (block != NULL) { + cl_core_local_block_set_block(core_local_block, NULL); + cl_block_list_insert_at_tail(&g_log.dirty_block_list, block); + } + block = cl_allocate_block(); + if (block == NULL) { + return false; + } + cl_core_local_block_set_block(core_local_block, block); + cl_block_enable_access(block); + return true; +} + +static cl_block* cl_get_block(void* record) { + uintptr_t p = (uintptr_t)((char*)record - g_log.buffer); + uintptr_t index = p >> CENSUS_LOG_2_MAX_RECORD_SIZE; + return &g_log.blocks[index]; +} + +// Gets the next block to read and tries to free 'prev' block (if not NULL). +// Returns NULL if reached the end. +static cl_block* cl_next_block_to_read(cl_block* prev) { + cl_block* block = NULL; + if (g_log.read_iterator_state == g_log.num_cores) { + // We are traversing dirty list; find the next dirty block. + if (prev != NULL) { + // Try to free the previous block if there is no unread data. This + // block + // may have unread data if previously incomplete record completed + // between + // read_next() calls. + block = prev->link.next->block; + if (cl_block_try_disable_access(prev, 0 /* do not discard data */)) { + cl_block_list_remove(&g_log.dirty_block_list, prev); + cl_block_list_insert_at_head(&g_log.free_block_list, prev); + } + } else { + block = cl_block_list_head(&g_log.dirty_block_list); + } + if (block != NULL) { + return block; + } + // We are done with the dirty list; moving on to core-local blocks. + } + while (g_log.read_iterator_state > 0) { + g_log.read_iterator_state--; + block = cl_core_local_block_get_block( + &g_log.core_local_blocks[g_log.read_iterator_state]); + if (block != NULL) { + return block; + } + } + return NULL; +} + +#define CL_LOG_2_MB 20 // 2^20 = 1MB + +// External functions: primary stats_log interface +void census_log_initialize(size_t size_in_mb, int discard_old_records) { + // Check cacheline alignment. + GPR_ASSERT(sizeof(cl_block) % GPR_CACHELINE_SIZE == 0); + GPR_ASSERT(sizeof(cl_core_local_block) % GPR_CACHELINE_SIZE == 0); + GPR_ASSERT(!g_log.initialized); + g_log.discard_old_records = discard_old_records; + g_log.num_cores = gpr_cpu_num_cores(); + // Ensure that we will not get any overflow in calaculating num_blocks + GPR_ASSERT(CL_LOG_2_MB >= CENSUS_LOG_2_MAX_RECORD_SIZE); + GPR_ASSERT(size_in_mb < 1000); + // Ensure at least 2x as many blocks as there are cores. + g_log.num_blocks = + (uint32_t)GPR_MAX(2 * g_log.num_cores, (size_in_mb << CL_LOG_2_MB) >> + CENSUS_LOG_2_MAX_RECORD_SIZE); + gpr_mu_init(&g_log.lock); + g_log.read_iterator_state = 0; + g_log.block_being_read = NULL; + g_log.core_local_blocks = (cl_core_local_block*)gpr_malloc_aligned( + g_log.num_cores * sizeof(cl_core_local_block), GPR_CACHELINE_SIZE_LOG); + memset(g_log.core_local_blocks, 0, + g_log.num_cores * sizeof(cl_core_local_block)); + g_log.blocks = (cl_block*)gpr_malloc_aligned( + g_log.num_blocks * sizeof(cl_block), GPR_CACHELINE_SIZE_LOG); + memset(g_log.blocks, 0, g_log.num_blocks * sizeof(cl_block)); + g_log.buffer = gpr_malloc(g_log.num_blocks * CENSUS_LOG_MAX_RECORD_SIZE); + memset(g_log.buffer, 0, g_log.num_blocks * CENSUS_LOG_MAX_RECORD_SIZE); + cl_block_list_initialize(&g_log.free_block_list); + cl_block_list_initialize(&g_log.dirty_block_list); + for (uint32_t i = 0; i < g_log.num_blocks; ++i) { + cl_block* block = g_log.blocks + i; + cl_block_initialize(block, g_log.buffer + (CENSUS_LOG_MAX_RECORD_SIZE * i)); + cl_block_try_disable_access(block, 1 /* discard data */); + cl_block_list_insert_at_tail(&g_log.free_block_list, block); + } + gpr_atm_rel_store(&g_log.out_of_space_count, 0); + g_log.initialized = 1; +} + +void census_log_shutdown(void) { + GPR_ASSERT(g_log.initialized); + gpr_mu_destroy(&g_log.lock); + gpr_free_aligned(g_log.core_local_blocks); + g_log.core_local_blocks = NULL; + gpr_free_aligned(g_log.blocks); + g_log.blocks = NULL; + gpr_free(g_log.buffer); + g_log.buffer = NULL; + g_log.initialized = 0; +} + +void* census_log_start_write(size_t size) { + // Used to bound number of times block allocation is attempted. + GPR_ASSERT(size > 0); + GPR_ASSERT(g_log.initialized); + if (size > CENSUS_LOG_MAX_RECORD_SIZE) { + return NULL; + } + uint32_t attempts_remaining = g_log.num_blocks; + uint32_t core_id = gpr_cpu_current_cpu(); + do { + void* record = NULL; + cl_block* block = + cl_core_local_block_get_block(&g_log.core_local_blocks[core_id]); + if (block && (record = cl_block_start_write(block, size))) { + return record; + } + // Need to allocate a new block. We are here if: + // - No block associated with the core OR + // - Write in-progress on the block OR + // - block is out of space + gpr_mu_lock(&g_log.lock); + bool allocated = cl_allocate_core_local_block(core_id, block); + gpr_mu_unlock(&g_log.lock); + if (!allocated) { + gpr_atm_no_barrier_fetch_add(&g_log.out_of_space_count, 1); + return NULL; + } + } while (attempts_remaining--); + // Give up. + gpr_atm_no_barrier_fetch_add(&g_log.out_of_space_count, 1); + return NULL; +} + +void census_log_end_write(void* record, size_t bytes_written) { + GPR_ASSERT(g_log.initialized); + cl_block_end_write(cl_get_block(record), bytes_written); +} + +void census_log_init_reader(void) { + GPR_ASSERT(g_log.initialized); + gpr_mu_lock(&g_log.lock); + // If a block is locked for reading unlock it. + if (g_log.block_being_read != NULL) { + cl_block_end_read(g_log.block_being_read); + g_log.block_being_read = NULL; + } + g_log.read_iterator_state = g_log.num_cores; + gpr_mu_unlock(&g_log.lock); +} + +const void* census_log_read_next(size_t* bytes_available) { + GPR_ASSERT(g_log.initialized); + gpr_mu_lock(&g_log.lock); + if (g_log.block_being_read != NULL) { + cl_block_end_read(g_log.block_being_read); + } + do { + g_log.block_being_read = cl_next_block_to_read(g_log.block_being_read); + if (g_log.block_being_read != NULL) { + void* record = + cl_block_start_read(g_log.block_being_read, bytes_available); + if (record != NULL) { + gpr_mu_unlock(&g_log.lock); + return record; + } + } + } while (g_log.block_being_read != NULL); + gpr_mu_unlock(&g_log.lock); + return NULL; +} + +size_t census_log_remaining_space(void) { + GPR_ASSERT(g_log.initialized); + size_t space = 0; + gpr_mu_lock(&g_log.lock); + if (g_log.discard_old_records) { + // Remaining space is not meaningful; just return the entire log space. + space = g_log.num_blocks << CENSUS_LOG_2_MAX_RECORD_SIZE; + } else { + GPR_ASSERT(g_log.free_block_list.count >= 0); + space = (size_t)g_log.free_block_list.count * CENSUS_LOG_MAX_RECORD_SIZE; + } + gpr_mu_unlock(&g_log.lock); + return space; +} + +int64_t census_log_out_of_space_count(void) { + GPR_ASSERT(g_log.initialized); + return gpr_atm_acq_load(&g_log.out_of_space_count); +} diff --git a/src/core/lib/census/mlog.h b/src/core/lib/census/mlog.h new file mode 100644 index 0000000000..7fbdeda986 --- /dev/null +++ b/src/core/lib/census/mlog.h @@ -0,0 +1,95 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* A very fast in-memory log, optimized for multiple writers. */ + +#ifndef GRPC_CORE_LIB_CENSUS_MLOG_H +#define GRPC_CORE_LIB_CENSUS_MLOG_H + +#include <grpc/support/port_platform.h> +#include <stddef.h> + +/* Maximum record size, in bytes. */ +#define CENSUS_LOG_2_MAX_RECORD_SIZE 14 /* 2^14 = 16KB */ +#define CENSUS_LOG_MAX_RECORD_SIZE (1 << CENSUS_LOG_2_MAX_RECORD_SIZE) + +/* Initialize the statistics logging subsystem with the given log size. A log + size of 0 will result in the smallest possible log for the platform + (approximately CENSUS_LOG_MAX_RECORD_SIZE * gpr_cpu_num_cores()). If + discard_old_records is non-zero, then new records will displace older ones + when the log is full. This function must be called before any other + census_log functions. +*/ +void census_log_initialize(size_t size_in_mb, int discard_old_records); + +/* Shutdown the logging subsystem. Caller must ensure that: + - no in progress or future call to any census_log functions + - no incomplete records +*/ +void census_log_shutdown(void); + +/* Allocates and returns a 'size' bytes record and marks it in use. A + subsequent census_log_end_write() marks the record complete. The + 'bytes_written' census_log_end_write() argument must be <= + 'size'. Returns NULL if out-of-space AND: + - log is configured to keep old records OR + - all blocks are pinned by incomplete records. +*/ +void* census_log_start_write(size_t size); + +void census_log_end_write(void* record, size_t bytes_written); + +void census_log_init_reader(void); + +/* census_log_read_next() iterates over blocks with data and for each block + returns a pointer to the first unread byte. The number of bytes that can be + read are returned in 'bytes_available'. Reader is expected to read all + available data. Reading the data consumes it i.e. it cannot be read again. + census_log_read_next() returns NULL if the end is reached i.e last block + is read. census_log_init_reader() starts the iteration or aborts the + current iteration. +*/ +const void* census_log_read_next(size_t* bytes_available); + +/* Returns estimated remaining space across all blocks, in bytes. If log is + configured to discard old records, returns total log space. Otherwise, + returns space available in empty blocks (partially filled blocks are + treated as full). +*/ +size_t census_log_remaining_space(void); + +/* Returns the number of times gprc_stats_log_start_write() failed due to + out-of-space. */ +int64_t census_log_out_of_space_count(void); + +#endif /* GRPC_CORE_LIB_CENSUS_MLOG_H */ diff --git a/src/core/lib/census/operation.c b/src/core/lib/census/operation.c new file mode 100644 index 0000000000..315f9c3534 --- /dev/null +++ b/src/core/lib/census/operation.c @@ -0,0 +1,63 @@ +/* + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/census.h> + +/* TODO(aveitch): These are all placeholder implementations. */ + +census_timestamp census_start_rpc_op_timestamp(void) { + census_timestamp ct; + /* TODO(aveitch): assumes gpr_timespec implementation of census_timestamp. */ + ct.ts = gpr_now(GPR_CLOCK_MONOTONIC); + return ct; +} + +census_context *census_start_client_rpc_op( + const census_context *context, int64_t rpc_name_id, + const census_rpc_name_info *rpc_name_info, const char *peer, int trace_mask, + const census_timestamp *start_time) { + return NULL; +} + +census_context *census_start_server_rpc_op( + const char *buffer, int64_t rpc_name_id, + const census_rpc_name_info *rpc_name_info, const char *peer, int trace_mask, + census_timestamp *start_time) { + return NULL; +} + +census_context *census_start_op(census_context *context, const char *family, + const char *name, int trace_mask) { + return NULL; +} + +void census_end_op(census_context *context, int status) {} diff --git a/src/core/lib/census/placeholders.c b/src/core/lib/census/placeholders.c new file mode 100644 index 0000000000..fe23d13971 --- /dev/null +++ b/src/core/lib/census/placeholders.c @@ -0,0 +1,109 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/census.h> + +#include <grpc/support/log.h> + +/* Placeholders for the pending APIs */ + +int census_get_trace_record(census_trace_record *trace_record) { + (void)trace_record; + abort(); +} + +void census_record_values(census_context *context, census_value *values, + size_t nvalues) { + (void)context; + (void)values; + (void)nvalues; + abort(); +} + +void census_set_rpc_client_peer(census_context *context, const char *peer) { + (void)context; + (void)peer; + abort(); +} + +void census_trace_scan_end() { abort(); } + +int census_trace_scan_start(int consume) { + (void)consume; + abort(); +} + +const census_aggregation *census_view_aggregrations(const census_view *view) { + (void)view; + abort(); +} + +census_view *census_view_create(uint32_t metric_id, const census_context *tags, + const census_aggregation *aggregations, + size_t naggregations) { + (void)metric_id; + (void)tags; + (void)aggregations; + (void)naggregations; + abort(); +} + +const census_context *census_view_tags(const census_view *view) { + (void)view; + abort(); +} + +void census_view_delete(census_view *view) { + (void)view; + abort(); +} + +const census_view_data *census_view_get_data(const census_view *view) { + (void)view; + abort(); +} + +size_t census_view_metric(const census_view *view) { + (void)view; + abort(); +} + +size_t census_view_naggregations(const census_view *view) { + (void)view; + abort(); +} + +void census_view_reset(census_view *view) { + (void)view; + abort(); +} diff --git a/src/core/lib/census/rpc_metric_id.h b/src/core/lib/census/rpc_metric_id.h new file mode 100644 index 0000000000..aad0588fb3 --- /dev/null +++ b/src/core/lib/census/rpc_metric_id.h @@ -0,0 +1,51 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_CENSUS_RPC_METRIC_ID_H +#define GRPC_CORE_LIB_CENSUS_RPC_METRIC_ID_H + +/* Metric ID's used for RPC measurements. */ +/* Count of client requests sent. */ +#define CENSUS_METRIC_RPC_CLIENT_REQUESTS ((uint32_t)0) +/* Count of server requests sent. */ +#define CENSUS_METRIC_RPC_SERVER_REQUESTS ((uint32_t)1) +/* Client error counts. */ +#define CENSUS_METRIC_RPC_CLIENT_ERRORS ((uint32_t)2) +/* Server error counts. */ +#define CENSUS_METRIC_RPC_SERVER_ERRORS ((uint32_t)3) +/* Client side request latency. */ +#define CENSUS_METRIC_RPC_CLIENT_LATENCY ((uint32_t)4) +/* Server side request latency. */ +#define CENSUS_METRIC_RPC_SERVER_LATENCY ((uint32_t)5) + +#endif /* GRPC_CORE_LIB_CENSUS_RPC_METRIC_ID_H */ diff --git a/src/core/lib/census/tracing.c b/src/core/lib/census/tracing.c new file mode 100644 index 0000000000..e508996af3 --- /dev/null +++ b/src/core/lib/census/tracing.c @@ -0,0 +1,45 @@ +/* + * + * Copyright 2015-2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/census.h> + +/* TODO(aveitch): These are all placeholder implementations. */ + +int census_trace_mask(const census_context *context) { + return CENSUS_TRACE_MASK_NONE; +} + +void census_set_trace_mask(int trace_mask) {} + +void census_trace_print(census_context *context, uint32_t type, + const char *buffer, size_t n) {} |