Merge branch 'master' of https://github.com/grpc/grpc into serialization
ncteisen
7 years ago
commit
8db7788d46
165 changed files with 11854 additions and 11364 deletions
@ -0,0 +1,63 @@ |
||||
# gRPC Concepts Overview |
||||
|
||||
Remote Procedure Calls (RPCs) provide a useful abstraction for building |
||||
distributed applications and services. The libraries in this repository |
||||
provide a concrete implementation of the gRPC protocol, layered over HTTP/2. |
||||
These libraries enable communication between clients and servers using any |
||||
combination of the supported languages. |
||||
|
||||
|
||||
## Interface |
||||
|
||||
Developers using gRPC start with a language agnostic description of an RPC service (a collection |
||||
of methods). From this description, gRPC will generate client and server side interfaces |
||||
in any of the supported languages. The server implements |
||||
the service interface, which can be remotely invoked by the client interface. |
||||
|
||||
By default, gRPC uses [Protocol Buffers](https://github.com/google/protobuf) as the |
||||
Interface Definition Language (IDL) for describing both the service interface |
||||
and the structure of the payload messages. It is possible to use other |
||||
alternatives if desired. |
||||
|
||||
### Invoking & handling remote calls |
||||
Starting from an interface definition in a .proto file, gRPC provides |
||||
Protocol Compiler plugins that generate Client- and Server-side APIs. |
||||
gRPC users call into these APIs on the Client side and implement |
||||
the corresponding API on the server side. |
||||
|
||||
#### Synchronous vs. asynchronous |
||||
Synchronous RPC calls, that block until a response arrives from the server, are |
||||
the closest approximation to the abstraction of a procedure call that RPC |
||||
aspires to. |
||||
|
||||
On the other hand, networks are inherently asynchronous and in many scenarios, |
||||
it is desirable to have the ability to start RPCs without blocking the current |
||||
thread. |
||||
|
||||
The gRPC programming surface in most languages comes in both synchronous and |
||||
asynchronous flavors. |
||||
|
||||
|
||||
## Streaming |
||||
|
||||
gRPC supports streaming semantics, where either the client or the server (or both) |
||||
send a stream of messages on a single RPC call. The most general case is |
||||
Bidirectional Streaming where a single gRPC call establishes a stream in which both |
||||
the client and the server can send a stream of messages to each other. The streamed |
||||
messages are delivered in the order they were sent. |
||||
|
||||
|
||||
# Protocol |
||||
|
||||
The [gRPC protocol](doc/PROTOCOL-HTTP2.md) specifies the abstract requirements for communication between |
||||
clients and servers. A concrete embedding over HTTP/2 completes the picture by |
||||
fleshing out the details of each of the required operations. |
||||
|
||||
## Abstract gRPC protocol |
||||
A gRPC call comprises of a bidirectional stream of messages, initiated by the client. In the client-to-server direction, this stream begins with a mandatory `Call Header`, followed by optional `Initial-Metadata`, followed by zero or more `Payload Messages`. The server-to-client direction contains an optional `Initial-Metadata`, followed by zero or more `Payload Messages` terminated with a mandatory `Status` and optional `Status-Metadata` (a.k.a.,`Trailing-Metadata`). |
||||
|
||||
## Implementation over HTTP/2 |
||||
The abstract protocol defined above is implemented over [HTTP/2](https://http2.github.io/). gRPC bidirectional streams are mapped to HTTP/2 streams. The contents of `Call Header` and `Initial Metadata` are sent as HTTP/2 headers and subject to HPACK compression. `Payload Messages` are serialized into a byte stream of length prefixed gRPC frames which are then fragmented into HTTP/2 frames at the sender and reassembled at the receiver. `Status` and `Trailing-Metadata` are sent as HTTP/2 trailing headers (a.k.a., trailers). |
||||
|
||||
## Flow Control |
||||
gRPC uses the flow control mechanism in HTTP/2. This enables fine-grained control of memory used for buffering in-flight messages. |
||||
@ -0,0 +1,6 @@ |
||||
# gRPC Basics: C++ sample code |
||||
|
||||
The files in this folder are the samples used in [gRPC Basics: C++][], |
||||
a detailed tutorial for using gRPC in C++. |
||||
|
||||
[gRPC Basics: C++]:https://grpc.io/docs/tutorials/basic/c.html |
||||
@ -0,0 +1,19 @@ |
||||
/* Automatically generated nanopb constant definitions */ |
||||
/* Generated by nanopb-0.3.7-dev */ |
||||
|
||||
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/google/protobuf/duration.pb.h" |
||||
/* @@protoc_insertion_point(includes) */ |
||||
#if PB_PROTO_HEADER_VERSION != 30 |
||||
#error Regenerate this file with the current version of nanopb generator. |
||||
#endif |
||||
|
||||
|
||||
|
||||
const pb_field_t google_protobuf_Duration_fields[3] = { |
||||
PB_FIELD( 1, INT64 , OPTIONAL, STATIC , FIRST, google_protobuf_Duration, seconds, seconds, 0), |
||||
PB_FIELD( 2, INT32 , OPTIONAL, STATIC , OTHER, google_protobuf_Duration, nanos, seconds, 0), |
||||
PB_LAST_FIELD |
||||
}; |
||||
|
||||
|
||||
/* @@protoc_insertion_point(eof) */ |
||||
@ -0,0 +1,54 @@ |
||||
/* Automatically generated nanopb header */ |
||||
/* Generated by nanopb-0.3.7-dev */ |
||||
|
||||
#ifndef PB_GOOGLE_PROTOBUF_DURATION_PB_H_INCLUDED |
||||
#define PB_GOOGLE_PROTOBUF_DURATION_PB_H_INCLUDED |
||||
#include "pb.h" |
||||
/* @@protoc_insertion_point(includes) */ |
||||
#if PB_PROTO_HEADER_VERSION != 30 |
||||
#error Regenerate this file with the current version of nanopb generator. |
||||
#endif |
||||
|
||||
#ifdef __cplusplus |
||||
extern "C" { |
||||
#endif |
||||
|
||||
/* Struct definitions */ |
||||
typedef struct _google_protobuf_Duration { |
||||
bool has_seconds; |
||||
int64_t seconds; |
||||
bool has_nanos; |
||||
int32_t nanos; |
||||
/* @@protoc_insertion_point(struct:google_protobuf_Duration) */ |
||||
} google_protobuf_Duration; |
||||
|
||||
/* Default values for struct fields */ |
||||
|
||||
/* Initializer values for message structs */ |
||||
#define google_protobuf_Duration_init_default {false, 0, false, 0} |
||||
#define google_protobuf_Duration_init_zero {false, 0, false, 0} |
||||
|
||||
/* Field tags (for use in manual encoding/decoding) */ |
||||
#define google_protobuf_Duration_seconds_tag 1 |
||||
#define google_protobuf_Duration_nanos_tag 2 |
||||
|
||||
/* Struct field encoding specification for nanopb */ |
||||
extern const pb_field_t google_protobuf_Duration_fields[3]; |
||||
|
||||
/* Maximum encoded size of messages (where known) */ |
||||
#define google_protobuf_Duration_size 22 |
||||
|
||||
/* Message IDs (where set with "msgid" option) */ |
||||
#ifdef PB_MSGID |
||||
|
||||
#define DURATION_MESSAGES \ |
||||
|
||||
|
||||
#endif |
||||
|
||||
#ifdef __cplusplus |
||||
} /* extern "C" */ |
||||
#endif |
||||
/* @@protoc_insertion_point(eof) */ |
||||
|
||||
#endif |
||||
@ -0,0 +1,19 @@ |
||||
/* Automatically generated nanopb constant definitions */ |
||||
/* Generated by nanopb-0.3.7-dev */ |
||||
|
||||
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/google/protobuf/timestamp.pb.h" |
||||
/* @@protoc_insertion_point(includes) */ |
||||
#if PB_PROTO_HEADER_VERSION != 30 |
||||
#error Regenerate this file with the current version of nanopb generator. |
||||
#endif |
||||
|
||||
|
||||
|
||||
const pb_field_t google_protobuf_Timestamp_fields[3] = { |
||||
PB_FIELD( 1, INT64 , OPTIONAL, STATIC , FIRST, google_protobuf_Timestamp, seconds, seconds, 0), |
||||
PB_FIELD( 2, INT32 , OPTIONAL, STATIC , OTHER, google_protobuf_Timestamp, nanos, seconds, 0), |
||||
PB_LAST_FIELD |
||||
}; |
||||
|
||||
|
||||
/* @@protoc_insertion_point(eof) */ |
||||
@ -0,0 +1,54 @@ |
||||
/* Automatically generated nanopb header */ |
||||
/* Generated by nanopb-0.3.7-dev */ |
||||
|
||||
#ifndef PB_GOOGLE_PROTOBUF_TIMESTAMP_PB_H_INCLUDED |
||||
#define PB_GOOGLE_PROTOBUF_TIMESTAMP_PB_H_INCLUDED |
||||
#include "pb.h" |
||||
/* @@protoc_insertion_point(includes) */ |
||||
#if PB_PROTO_HEADER_VERSION != 30 |
||||
#error Regenerate this file with the current version of nanopb generator. |
||||
#endif |
||||
|
||||
#ifdef __cplusplus |
||||
extern "C" { |
||||
#endif |
||||
|
||||
/* Struct definitions */ |
||||
typedef struct _google_protobuf_Timestamp { |
||||
bool has_seconds; |
||||
int64_t seconds; |
||||
bool has_nanos; |
||||
int32_t nanos; |
||||
/* @@protoc_insertion_point(struct:google_protobuf_Timestamp) */ |
||||
} google_protobuf_Timestamp; |
||||
|
||||
/* Default values for struct fields */ |
||||
|
||||
/* Initializer values for message structs */ |
||||
#define google_protobuf_Timestamp_init_default {false, 0, false, 0} |
||||
#define google_protobuf_Timestamp_init_zero {false, 0, false, 0} |
||||
|
||||
/* Field tags (for use in manual encoding/decoding) */ |
||||
#define google_protobuf_Timestamp_seconds_tag 1 |
||||
#define google_protobuf_Timestamp_nanos_tag 2 |
||||
|
||||
/* Struct field encoding specification for nanopb */ |
||||
extern const pb_field_t google_protobuf_Timestamp_fields[3]; |
||||
|
||||
/* Maximum encoded size of messages (where known) */ |
||||
#define google_protobuf_Timestamp_size 22 |
||||
|
||||
/* Message IDs (where set with "msgid" option) */ |
||||
#ifdef PB_MSGID |
||||
|
||||
#define TIMESTAMP_MESSAGES \ |
||||
|
||||
|
||||
#endif |
||||
|
||||
#ifdef __cplusplus |
||||
} /* extern "C" */ |
||||
#endif |
||||
/* @@protoc_insertion_point(eof) */ |
||||
|
||||
#endif |
||||
@ -0,0 +1,108 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_REGISTERED_OPENCENSUS_OBJECTS_H |
||||
#define GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_REGISTERED_OPENCENSUS_OBJECTS_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "opencensus/stats/stats.h" |
||||
|
||||
#include "src/cpp/server/load_reporter/constants.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
// Measures.
|
||||
|
||||
::opencensus::stats::MeasureInt64 MeasureStartCount() { |
||||
static const ::opencensus::stats::MeasureInt64 start_count = |
||||
::opencensus::stats::MeasureInt64::Register( |
||||
kMeasureStartCount, kMeasureStartCount, kMeasureStartCount); |
||||
return start_count; |
||||
} |
||||
|
||||
::opencensus::stats::MeasureInt64 MeasureEndCount() { |
||||
static const ::opencensus::stats::MeasureInt64 end_count = |
||||
::opencensus::stats::MeasureInt64::Register( |
||||
kMeasureEndCount, kMeasureEndCount, kMeasureEndCount); |
||||
return end_count; |
||||
} |
||||
|
||||
::opencensus::stats::MeasureInt64 MeasureEndBytesSent() { |
||||
static const ::opencensus::stats::MeasureInt64 end_bytes_sent = |
||||
::opencensus::stats::MeasureInt64::Register( |
||||
kMeasureEndBytesSent, kMeasureEndBytesSent, kMeasureEndBytesSent); |
||||
return end_bytes_sent; |
||||
} |
||||
|
||||
::opencensus::stats::MeasureInt64 MeasureEndBytesReceived() { |
||||
static const ::opencensus::stats::MeasureInt64 end_bytes_received = |
||||
::opencensus::stats::MeasureInt64::Register(kMeasureEndBytesReceived, |
||||
kMeasureEndBytesReceived, |
||||
kMeasureEndBytesReceived); |
||||
return end_bytes_received; |
||||
} |
||||
|
||||
::opencensus::stats::MeasureInt64 MeasureEndLatencyMs() { |
||||
static const ::opencensus::stats::MeasureInt64 end_latency_ms = |
||||
::opencensus::stats::MeasureInt64::Register( |
||||
kMeasureEndLatencyMs, kMeasureEndLatencyMs, kMeasureEndLatencyMs); |
||||
return end_latency_ms; |
||||
} |
||||
|
||||
::opencensus::stats::MeasureDouble MeasureOtherCallMetric() { |
||||
static const ::opencensus::stats::MeasureDouble other_call_metric = |
||||
::opencensus::stats::MeasureDouble::Register(kMeasureOtherCallMetric, |
||||
kMeasureOtherCallMetric, |
||||
kMeasureOtherCallMetric); |
||||
return other_call_metric; |
||||
} |
||||
|
||||
// Tags.
|
||||
|
||||
opencensus::stats::TagKey TagKeyToken() { |
||||
static const auto token = opencensus::stats::TagKey::Register(kTagKeyToken); |
||||
return token; |
||||
} |
||||
|
||||
opencensus::stats::TagKey TagKeyHost() { |
||||
static const auto token = opencensus::stats::TagKey::Register(kTagKeyHost); |
||||
return token; |
||||
} |
||||
opencensus::stats::TagKey TagKeyUserId() { |
||||
static const auto token = opencensus::stats::TagKey::Register(kTagKeyUserId); |
||||
return token; |
||||
} |
||||
|
||||
opencensus::stats::TagKey TagKeyStatus() { |
||||
static const auto token = opencensus::stats::TagKey::Register(kTagKeyStatus); |
||||
return token; |
||||
} |
||||
|
||||
opencensus::stats::TagKey TagKeyMetricName() { |
||||
static const auto token = |
||||
opencensus::stats::TagKey::Register(kTagKeyMetricName); |
||||
return token; |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_REGISTERED_OPENCENSUS_OBJECTS_H \ |
||||
*/ |
||||
@ -1,71 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <limits.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/load_reporting.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/sync.h> |
||||
|
||||
#include "src/core/ext/filters/load_reporting/server_load_reporting_filter.h" |
||||
#include "src/core/ext/filters/load_reporting/server_load_reporting_plugin.h" |
||||
#include "src/core/lib/channel/channel_stack_builder.h" |
||||
#include "src/core/lib/slice/slice_internal.h" |
||||
#include "src/core/lib/surface/call.h" |
||||
#include "src/core/lib/surface/channel_init.h" |
||||
|
||||
static bool is_load_reporting_enabled(const grpc_channel_args* a) { |
||||
return grpc_channel_arg_get_bool( |
||||
grpc_channel_args_find(a, GRPC_ARG_ENABLE_LOAD_REPORTING), false); |
||||
} |
||||
|
||||
static bool maybe_add_server_load_reporting_filter( |
||||
grpc_channel_stack_builder* builder, void* arg) { |
||||
const grpc_channel_args* args = |
||||
grpc_channel_stack_builder_get_channel_arguments(builder); |
||||
const grpc_channel_filter* filter = |
||||
static_cast<const grpc_channel_filter*>(arg); |
||||
grpc_channel_stack_builder_iterator* it = |
||||
grpc_channel_stack_builder_iterator_find(builder, filter->name); |
||||
const bool already_has_load_reporting_filter = |
||||
!grpc_channel_stack_builder_iterator_is_end(it); |
||||
grpc_channel_stack_builder_iterator_destroy(it); |
||||
if (is_load_reporting_enabled(args) && !already_has_load_reporting_filter) { |
||||
return grpc_channel_stack_builder_prepend_filter(builder, filter, nullptr, |
||||
nullptr); |
||||
} |
||||
return true; |
||||
} |
||||
|
||||
grpc_arg grpc_load_reporting_enable_arg() { |
||||
return grpc_channel_arg_integer_create((char*)GRPC_ARG_ENABLE_LOAD_REPORTING, |
||||
1); |
||||
} |
||||
|
||||
/* Plugin registration */ |
||||
|
||||
void grpc_server_load_reporting_plugin_init(void) { |
||||
grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX, |
||||
maybe_add_server_load_reporting_filter, |
||||
(void*)&grpc_server_load_reporting_filter); |
||||
} |
||||
|
||||
void grpc_server_load_reporting_plugin_shutdown() {} |
||||
@ -1,61 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_PLUGIN_H |
||||
#define GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_PLUGIN_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <grpc/impl/codegen/grpc_types.h> |
||||
|
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
|
||||
/** Identifiers for the invocation point of the users LR callback */ |
||||
typedef enum grpc_load_reporting_source { |
||||
GRPC_LR_POINT_UNKNOWN = 0, |
||||
GRPC_LR_POINT_CHANNEL_CREATION, |
||||
GRPC_LR_POINT_CHANNEL_DESTRUCTION, |
||||
GRPC_LR_POINT_CALL_CREATION, |
||||
GRPC_LR_POINT_CALL_DESTRUCTION |
||||
} grpc_load_reporting_source; |
||||
|
||||
/** Call information to be passed to the provided LR callback. */ |
||||
typedef struct grpc_load_reporting_call_data { |
||||
const grpc_load_reporting_source source; /**< point of last data update. */ |
||||
|
||||
/** Unique identifier for the channel associated with the data */ |
||||
intptr_t channel_id; |
||||
|
||||
/** Unique identifier for the call associated with the data. If the call
|
||||
* hasn't been created yet, it'll have a value of zero. */ |
||||
intptr_t call_id; |
||||
|
||||
/** Only valid when \a source is \a GRPC_LR_POINT_CALL_DESTRUCTION, that is,
|
||||
* once the call has completed */ |
||||
const grpc_call_final_info* final_info; |
||||
|
||||
const char* initial_md_string; /**< value string for LR's initial md key */ |
||||
const char* trailing_md_string; /**< value string for LR's trailing md key */ |
||||
const char* method_name; /**< Corresponds to :path header */ |
||||
} grpc_load_reporting_call_data; |
||||
|
||||
/** Return a \a grpc_arg enabling load reporting */ |
||||
grpc_arg grpc_load_reporting_enable_arg(); |
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_PLUGIN_H \ |
||||
*/ |
||||
@ -0,0 +1,188 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/impl/codegen/port_platform.h> |
||||
|
||||
#include "src/core/lib/channel/channelz.h" |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <stdio.h> |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
|
||||
#include "src/core/lib/channel/channelz_registry.h" |
||||
#include "src/core/lib/channel/status_util.h" |
||||
#include "src/core/lib/gpr/string.h" |
||||
#include "src/core/lib/gpr/useful.h" |
||||
#include "src/core/lib/gprpp/memory.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/slice/slice_internal.h" |
||||
#include "src/core/lib/surface/channel.h" |
||||
#include "src/core/lib/transport/connectivity_state.h" |
||||
#include "src/core/lib/transport/error_utils.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace channelz { |
||||
|
||||
namespace { |
||||
|
||||
// TODO(ncteisen): move this function to a common helper location.
|
||||
//
|
||||
// returns an allocated string that represents tm according to RFC-3339, and,
|
||||
// more specifically, follows:
|
||||
// https://developers.google.com/protocol-buffers/docs/proto3#json
|
||||
//
|
||||
// "Uses RFC 3339, where generated output will always be Z-normalized and uses
|
||||
// 0, 3, 6 or 9 fractional digits."
|
||||
char* fmt_time(gpr_timespec tm) { |
||||
char time_buffer[35]; |
||||
char ns_buffer[11]; // '.' + 9 digits of precision
|
||||
struct tm* tm_info = localtime((const time_t*)&tm.tv_sec); |
||||
strftime(time_buffer, sizeof(time_buffer), "%Y-%m-%dT%H:%M:%S", tm_info); |
||||
snprintf(ns_buffer, 11, ".%09d", tm.tv_nsec); |
||||
// This loop trims off trailing zeros by inserting a null character that the
|
||||
// right point. We iterate in chunks of three because we want 0, 3, 6, or 9
|
||||
// fractional digits.
|
||||
for (int i = 7; i >= 1; i -= 3) { |
||||
if (ns_buffer[i] == '0' && ns_buffer[i + 1] == '0' && |
||||
ns_buffer[i + 2] == '0') { |
||||
ns_buffer[i] = '\0'; |
||||
// Edge case in which all fractional digits were 0.
|
||||
if (i == 1) { |
||||
ns_buffer[0] = '\0'; |
||||
} |
||||
} else { |
||||
break; |
||||
} |
||||
} |
||||
char* full_time_str; |
||||
gpr_asprintf(&full_time_str, "%s%sZ", time_buffer, ns_buffer); |
||||
return full_time_str; |
||||
} |
||||
|
||||
// TODO(ncteisen); move this to json library
|
||||
grpc_json* add_num_str(grpc_json* parent, grpc_json* it, const char* name, |
||||
int64_t num) { |
||||
char* num_str; |
||||
gpr_asprintf(&num_str, "%" PRId64, num); |
||||
return grpc_json_create_child(it, parent, name, num_str, GRPC_JSON_STRING, |
||||
true); |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes) |
||||
: channel_(channel), target_(nullptr), channel_uuid_(-1) { |
||||
trace_.Init(channel_tracer_max_nodes); |
||||
target_ = UniquePtr<char>(grpc_channel_get_target(channel_)); |
||||
channel_uuid_ = ChannelzRegistry::Register(this); |
||||
gpr_atm_no_barrier_store(&last_call_started_millis_, |
||||
(gpr_atm)ExecCtx::Get()->Now()); |
||||
} |
||||
|
||||
ChannelNode::~ChannelNode() { |
||||
trace_.Destroy(); |
||||
ChannelzRegistry::Unregister(channel_uuid_); |
||||
} |
||||
|
||||
void ChannelNode::RecordCallStarted() { |
||||
gpr_atm_no_barrier_fetch_add(&calls_started_, (gpr_atm)1); |
||||
gpr_atm_no_barrier_store(&last_call_started_millis_, |
||||
(gpr_atm)ExecCtx::Get()->Now()); |
||||
} |
||||
|
||||
grpc_connectivity_state ChannelNode::GetConnectivityState() { |
||||
if (channel_ == nullptr) { |
||||
return GRPC_CHANNEL_SHUTDOWN; |
||||
} else { |
||||
// TODO(ncteisen): re-enable this once we have cleaned up all of the
|
||||
// internal dependency issues.
|
||||
// return grpc_channel_check_connectivity_state(channel_, false);
|
||||
return GRPC_CHANNEL_IDLE; |
||||
} |
||||
} |
||||
|
||||
char* ChannelNode::RenderJSON() { |
||||
// We need to track these three json objects to build our object
|
||||
grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); |
||||
grpc_json* json = top_level_json; |
||||
grpc_json* json_iterator = nullptr; |
||||
// create and fill the ref child
|
||||
json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr, |
||||
GRPC_JSON_OBJECT, false); |
||||
json = json_iterator; |
||||
json_iterator = nullptr; |
||||
json_iterator = add_num_str(json, json_iterator, "channelId", channel_uuid_); |
||||
// reset json iterators to top level object
|
||||
json = top_level_json; |
||||
json_iterator = nullptr; |
||||
// create and fill the data child.
|
||||
grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr, |
||||
GRPC_JSON_OBJECT, false); |
||||
json = data; |
||||
json_iterator = nullptr; |
||||
// create and fill the connectivity state child.
|
||||
grpc_connectivity_state connectivity_state = GetConnectivityState(); |
||||
json_iterator = grpc_json_create_child(json_iterator, json, "state", nullptr, |
||||
GRPC_JSON_OBJECT, false); |
||||
json = json_iterator; |
||||
grpc_json_create_child(nullptr, json, "state", |
||||
grpc_connectivity_state_name(connectivity_state), |
||||
GRPC_JSON_STRING, false); |
||||
// reset the parent to be the data object.
|
||||
json = data; |
||||
json_iterator = grpc_json_create_child( |
||||
json_iterator, json, "target", target_.get(), GRPC_JSON_STRING, false); |
||||
// fill in the channel trace if applicable
|
||||
grpc_json* trace = trace_->RenderJSON(); |
||||
if (trace != nullptr) { |
||||
// we manuall link up and fill the child since it was created for us in
|
||||
// ChannelTrace::RenderJSON
|
||||
json_iterator = grpc_json_link_child(json, trace, json_iterator); |
||||
trace->parent = json; |
||||
trace->value = nullptr; |
||||
trace->key = "trace"; |
||||
trace->owns_value = false; |
||||
} |
||||
// reset the parent to be the data object.
|
||||
json = data; |
||||
json_iterator = nullptr; |
||||
// We use -1 as sentinel values since proto default value for integers is
|
||||
// zero, and the confuses the parser into thinking the value weren't present
|
||||
json_iterator = |
||||
add_num_str(json, json_iterator, "callsStarted", calls_started_); |
||||
json_iterator = |
||||
add_num_str(json, json_iterator, "callsSucceeded", calls_succeeded_); |
||||
json_iterator = |
||||
add_num_str(json, json_iterator, "callsFailed", calls_failed_); |
||||
gpr_timespec ts = |
||||
grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME); |
||||
json_iterator = |
||||
grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp", |
||||
fmt_time(ts), GRPC_JSON_STRING, true); |
||||
// render and return the over json object
|
||||
char* json_str = grpc_json_dump_to_string(top_level_json, 0); |
||||
grpc_json_destroy(top_level_json); |
||||
return json_str; |
||||
} |
||||
|
||||
} // namespace channelz
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,85 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_LIB_CHANNEL_CHANNELZ_H |
||||
#define GRPC_CORE_LIB_CHANNEL_CHANNELZ_H |
||||
|
||||
#include <grpc/impl/codegen/port_platform.h> |
||||
|
||||
#include <grpc/grpc.h> |
||||
|
||||
#include "src/core/lib/channel/channel_trace.h" |
||||
#include "src/core/lib/gprpp/manual_constructor.h" |
||||
#include "src/core/lib/gprpp/ref_counted.h" |
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" |
||||
#include "src/core/lib/json/json.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace channelz { |
||||
|
||||
namespace testing { |
||||
class ChannelNodePeer; |
||||
} |
||||
|
||||
class ChannelNode : public RefCounted<ChannelNode> { |
||||
public: |
||||
ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes); |
||||
~ChannelNode(); |
||||
|
||||
void RecordCallStarted(); |
||||
void RecordCallFailed() { |
||||
gpr_atm_no_barrier_fetch_add(&calls_failed_, (gpr_atm(1))); |
||||
} |
||||
void RecordCallSucceeded() { |
||||
gpr_atm_no_barrier_fetch_add(&calls_succeeded_, (gpr_atm(1))); |
||||
} |
||||
|
||||
char* RenderJSON(); |
||||
|
||||
ChannelTrace* trace() { return trace_.get(); } |
||||
|
||||
void set_channel_destroyed() { |
||||
GPR_ASSERT(channel_ != nullptr); |
||||
channel_ = nullptr; |
||||
} |
||||
|
||||
intptr_t channel_uuid() { return channel_uuid_; } |
||||
|
||||
private: |
||||
// testing peer friend.
|
||||
friend class testing::ChannelNodePeer; |
||||
|
||||
// helper for getting connectivity state.
|
||||
grpc_connectivity_state GetConnectivityState(); |
||||
|
||||
grpc_channel* channel_ = nullptr; |
||||
UniquePtr<char> target_; |
||||
gpr_atm calls_started_ = 0; |
||||
gpr_atm calls_succeeded_ = 0; |
||||
gpr_atm calls_failed_ = 0; |
||||
gpr_atm last_call_started_millis_ = 0; |
||||
intptr_t channel_uuid_; |
||||
ManualConstructor<ChannelTrace> trace_; |
||||
}; |
||||
|
||||
} // namespace channelz
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif /* GRPC_CORE_LIB_CHANNEL_CHANNELZ_H */ |
||||
@ -0,0 +1,71 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_SRC_CPP_SERVER_LOAD_REPORTER_UTIL_H |
||||
#define GRPC_SRC_CPP_SERVER_LOAD_REPORTER_UTIL_H |
||||
|
||||
#include <grpc/impl/codegen/port_platform.h> |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
constexpr size_t kLbIdLength = 8; |
||||
constexpr size_t kIpv4AddressLength = 8; |
||||
constexpr size_t kIpv6AddressLength = 32; |
||||
|
||||
constexpr char kInvalidLbId[] = "<INVALID_LBID_238dsb234890rb>"; |
||||
|
||||
// Call statuses.
|
||||
|
||||
constexpr char kCallStatusOk[] = "OK"; |
||||
constexpr char kCallStatusServerError[] = "5XX"; |
||||
constexpr char kCallStatusClientError[] = "4XX"; |
||||
|
||||
// Tag keys.
|
||||
|
||||
constexpr char kTagKeyToken[] = "token"; |
||||
constexpr char kTagKeyHost[] = "host"; |
||||
constexpr char kTagKeyUserId[] = "user_id"; |
||||
constexpr char kTagKeyStatus[] = "status"; |
||||
constexpr char kTagKeyMetricName[] = "metric_name"; |
||||
|
||||
// Measure names.
|
||||
|
||||
constexpr char kMeasureStartCount[] = "grpc.io/lb/start_count"; |
||||
constexpr char kMeasureEndCount[] = "grpc.io/lb/end_count"; |
||||
constexpr char kMeasureEndBytesSent[] = "grpc.io/lb/bytes_sent"; |
||||
constexpr char kMeasureEndBytesReceived[] = "grpc.io/lb/bytes_received"; |
||||
constexpr char kMeasureEndLatencyMs[] = "grpc.io/lb/latency_ms"; |
||||
constexpr char kMeasureOtherCallMetric[] = "grpc.io/lb/other_call_metric"; |
||||
|
||||
// View names.
|
||||
|
||||
constexpr char kViewStartCount[] = "grpc.io/lb_view/start_count"; |
||||
constexpr char kViewEndCount[] = "grpc.io/lb_view/end_count"; |
||||
constexpr char kViewEndBytesSent[] = "grpc.io/lb_view/bytes_sent"; |
||||
constexpr char kViewEndBytesReceived[] = "grpc.io/lb_view/bytes_received"; |
||||
constexpr char kViewEndLatencyMs[] = "grpc.io/lb_view/latency_ms"; |
||||
constexpr char kViewOtherCallMetricCount[] = |
||||
"grpc.io/lb_view/other_call_metric_count"; |
||||
constexpr char kViewOtherCallMetricValue[] = |
||||
"grpc.io/lb_view/other_call_metric_value"; |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_SERVER_LOAD_REPORTER_UTIL_H
|
||||
@ -0,0 +1,36 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_SRC_CPP_SERVER_LOAD_REPORTER_GET_CPU_STATS_H |
||||
#define GRPC_SRC_CPP_SERVER_LOAD_REPORTER_GET_CPU_STATS_H |
||||
|
||||
#include <grpc/impl/codegen/port_platform.h> |
||||
|
||||
#include <utility> |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
// Reads the CPU stats (in a pair of busy and total numbers) from the system.
|
||||
// The units of the stats should be the same.
|
||||
std::pair<uint64_t, uint64_t> GetCpuStatsImpl(); |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_SERVER_LOAD_REPORTER_GET_CPU_STATS_H
|
||||
@ -0,0 +1,45 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_LINUX |
||||
|
||||
#include <cstdio> |
||||
|
||||
#include "src/cpp/server/load_reporter/get_cpu_stats.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
std::pair<uint64_t, uint64_t> GetCpuStatsImpl() { |
||||
uint64_t busy = 0, total = 0; |
||||
FILE* fp; |
||||
fp = fopen("/proc/stat", "r"); |
||||
uint64_t user, nice, system, idle; |
||||
fscanf(fp, "cpu %lu %lu %lu %lu", &user, &nice, &system, &idle); |
||||
fclose(fp); |
||||
busy = user + nice + system; |
||||
total = busy + idle; |
||||
return std::make_pair(busy, total); |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GPR_LINUX
|
||||
@ -0,0 +1,45 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_APPLE |
||||
|
||||
#include <mach/mach.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/get_cpu_stats.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
std::pair<uint64_t, uint64_t> GetCpuStatsImpl() { |
||||
uint64_t busy = 0, total = 0; |
||||
host_cpu_load_info_data_t cpuinfo; |
||||
mach_msg_type_number_t count = HOST_CPU_LOAD_INFO_COUNT; |
||||
if (host_statistics(mach_host_self(), HOST_CPU_LOAD_INFO, |
||||
(host_info_t)&cpuinfo, &count) == KERN_SUCCESS) { |
||||
for (int i = 0; i < CPU_STATE_MAX; i++) total += cpuinfo.cpu_ticks[i]; |
||||
busy = total - cpuinfo.cpu_ticks[CPU_STATE_IDLE]; |
||||
} |
||||
return std::make_pair(busy, total); |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GPR_APPLE
|
||||
@ -0,0 +1,40 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#if !defined(GPR_LINUX) && !defined(GPR_WINDOWS) && !defined(GPR_APPLE) |
||||
|
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/get_cpu_stats.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
std::pair<uint64_t, uint64_t> GetCpuStatsImpl() { |
||||
uint64_t busy = 0, total = 0; |
||||
gpr_log(GPR_ERROR, |
||||
"Platforms other than Linux, Windows, and MacOS are not supported."); |
||||
return std::make_pair(busy, total); |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // !defined(GPR_LINUX) && !defined(GPR_WINDOWS) && !defined(GPR_APPLE)
|
||||
@ -0,0 +1,55 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_WINDOWS |
||||
|
||||
#include <windows.h> |
||||
#include <cstdint> |
||||
|
||||
#include "src/cpp/server/load_reporter/get_cpu_stats.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
namespace { |
||||
|
||||
uint64_t FiletimeToInt(const FILETIME& ft) { |
||||
ULARGE_INTEGER i; |
||||
i.LowPart = ft.dwLowDateTime; |
||||
i.HighPart = ft.dwHighDateTime; |
||||
return i.QuadPart; |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
std::pair<uint64_t, uint64_t> GetCpuStatsImpl() { |
||||
uint64_t busy = 0, total = 0; |
||||
FILETIME idle, kernel, user; |
||||
if (GetSystemTimes(&idle, &kernel, &user) != 0) { |
||||
total = FiletimeToInt(kernel) + FiletimeToInt(user); |
||||
busy = total - FiletimeToInt(idle); |
||||
} |
||||
return std::make_pair(busy, total); |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GPR_WINDOWS
|
||||
@ -0,0 +1,498 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/impl/codegen/port_platform.h> |
||||
|
||||
#include <stdint.h> |
||||
#include <stdio.h> |
||||
#include <chrono> |
||||
#include <ctime> |
||||
|
||||
#include "src/cpp/server/load_reporter/constants.h" |
||||
#include "src/cpp/server/load_reporter/get_cpu_stats.h" |
||||
#include "src/cpp/server/load_reporter/load_reporter.h" |
||||
|
||||
#include "opencensus/stats/internal/set_aggregation_window.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
CpuStatsProvider::CpuStatsSample CpuStatsProviderDefaultImpl::GetCpuStats() { |
||||
return GetCpuStatsImpl(); |
||||
} |
||||
|
||||
CensusViewProvider::CensusViewProvider() |
||||
: tag_key_token_(::opencensus::stats::TagKey::Register(kTagKeyToken)), |
||||
tag_key_host_(::opencensus::stats::TagKey::Register(kTagKeyHost)), |
||||
tag_key_user_id_(::opencensus::stats::TagKey::Register(kTagKeyUserId)), |
||||
tag_key_status_(::opencensus::stats::TagKey::Register(kTagKeyStatus)), |
||||
tag_key_metric_name_( |
||||
::opencensus::stats::TagKey::Register(kTagKeyMetricName)) { |
||||
// One view related to starting a call.
|
||||
auto vd_start_count = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name(kViewStartCount) |
||||
.set_measure(kMeasureStartCount) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Sum()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.set_description( |
||||
"Delta count of calls started broken down by <token, host, " |
||||
"user_id>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_start_count); |
||||
view_descriptor_map_.emplace(kViewStartCount, vd_start_count); |
||||
// Four views related to ending a call.
|
||||
// If this view is set as Count of kMeasureEndBytesSent (in hope of saving one
|
||||
// measure), it's infeasible to prepare fake data for testing. That's because
|
||||
// the OpenCensus API to make up view data will add the input data as separate
|
||||
// measurements instead of setting the data values directly.
|
||||
auto vd_end_count = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name((kViewEndCount)) |
||||
.set_measure((kMeasureEndCount)) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Sum()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.add_column(tag_key_status_) |
||||
.set_description( |
||||
"Delta count of calls ended broken down by <token, host, " |
||||
"user_id, status>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_end_count); |
||||
view_descriptor_map_.emplace(kViewEndCount, vd_end_count); |
||||
auto vd_end_bytes_sent = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name((kViewEndBytesSent)) |
||||
.set_measure((kMeasureEndBytesSent)) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Sum()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.add_column(tag_key_status_) |
||||
.set_description( |
||||
"Delta sum of bytes sent broken down by <token, host, user_id, " |
||||
"status>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_end_bytes_sent); |
||||
view_descriptor_map_.emplace(kViewEndBytesSent, vd_end_bytes_sent); |
||||
auto vd_end_bytes_received = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name((kViewEndBytesReceived)) |
||||
.set_measure((kMeasureEndBytesReceived)) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Sum()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.add_column(tag_key_status_) |
||||
.set_description( |
||||
"Delta sum of bytes received broken down by <token, host, " |
||||
"user_id, status>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_end_bytes_received); |
||||
view_descriptor_map_.emplace(kViewEndBytesReceived, vd_end_bytes_received); |
||||
auto vd_end_latency_ms = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name((kViewEndLatencyMs)) |
||||
.set_measure((kMeasureEndLatencyMs)) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Sum()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.add_column(tag_key_status_) |
||||
.set_description( |
||||
"Delta sum of latency in ms broken down by <token, host, " |
||||
"user_id, status>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_end_latency_ms); |
||||
view_descriptor_map_.emplace(kViewEndLatencyMs, vd_end_latency_ms); |
||||
// Two views related to other call metrics.
|
||||
auto vd_metric_call_count = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name((kViewOtherCallMetricCount)) |
||||
.set_measure((kMeasureOtherCallMetric)) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Count()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.add_column(tag_key_metric_name_) |
||||
.set_description( |
||||
"Delta count of calls broken down by <token, host, user_id, " |
||||
"metric_name>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_metric_call_count); |
||||
view_descriptor_map_.emplace(kViewOtherCallMetricCount, vd_metric_call_count); |
||||
auto vd_metric_value = |
||||
::opencensus::stats::ViewDescriptor() |
||||
.set_name((kViewOtherCallMetricValue)) |
||||
.set_measure((kMeasureOtherCallMetric)) |
||||
.set_aggregation(::opencensus::stats::Aggregation::Sum()) |
||||
.add_column(tag_key_token_) |
||||
.add_column(tag_key_host_) |
||||
.add_column(tag_key_user_id_) |
||||
.add_column(tag_key_metric_name_) |
||||
.set_description( |
||||
"Delta sum of call metric value broken down " |
||||
"by <token, host, user_id, metric_name>."); |
||||
::opencensus::stats::SetAggregationWindow( |
||||
::opencensus::stats::AggregationWindow::Delta(), &vd_metric_value); |
||||
view_descriptor_map_.emplace(kViewOtherCallMetricValue, vd_metric_value); |
||||
} |
||||
|
||||
double CensusViewProvider::GetRelatedViewDataRowDouble( |
||||
const ViewDataMap& view_data_map, const char* view_name, |
||||
size_t view_name_len, const std::vector<grpc::string>& tag_values) { |
||||
auto it_vd = view_data_map.find(grpc::string(view_name, view_name_len)); |
||||
GPR_ASSERT(it_vd != view_data_map.end()); |
||||
auto it_row = it_vd->second.double_data().find(tag_values); |
||||
GPR_ASSERT(it_row != it_vd->second.double_data().end()); |
||||
return it_row->second; |
||||
} |
||||
|
||||
CensusViewProviderDefaultImpl::CensusViewProviderDefaultImpl() { |
||||
for (const auto& p : view_descriptor_map()) { |
||||
const grpc::string& view_name = p.first; |
||||
const ::opencensus::stats::ViewDescriptor& vd = p.second; |
||||
// We need to use pair's piecewise ctor here, otherwise the deleted copy
|
||||
// ctor of View will be called.
|
||||
view_map_.emplace(std::piecewise_construct, |
||||
std::forward_as_tuple(view_name), |
||||
std::forward_as_tuple(vd)); |
||||
} |
||||
} |
||||
|
||||
CensusViewProvider::ViewDataMap CensusViewProviderDefaultImpl::FetchViewData() { |
||||
gpr_log(GPR_DEBUG, "[CVP %p] Starts fetching Census view data.", this); |
||||
ViewDataMap view_data_map; |
||||
for (auto& p : view_map_) { |
||||
const grpc::string& view_name = p.first; |
||||
::opencensus::stats::View& view = p.second; |
||||
if (view.IsValid()) { |
||||
view_data_map.emplace(view_name, view.GetData()); |
||||
gpr_log(GPR_DEBUG, "[CVP %p] Fetched view data (view: %s).", this, |
||||
view_name.c_str()); |
||||
} else { |
||||
gpr_log( |
||||
GPR_DEBUG, |
||||
"[CVP %p] Can't fetch view data because view is invalid (view: %s).", |
||||
this, view_name.c_str()); |
||||
} |
||||
} |
||||
return view_data_map; |
||||
} |
||||
|
||||
grpc::string LoadReporter::GenerateLbId() { |
||||
while (true) { |
||||
if (next_lb_id_ > UINT32_MAX) { |
||||
gpr_log(GPR_ERROR, "[LR %p] The LB ID exceeds the max valid value!", |
||||
this); |
||||
return ""; |
||||
} |
||||
int64_t lb_id = next_lb_id_++; |
||||
// Overflow should never happen.
|
||||
GPR_ASSERT(lb_id >= 0); |
||||
// Convert to padded hex string for a 32-bit LB ID. E.g, "0000ca5b".
|
||||
char buf[kLbIdLength + 1]; |
||||
snprintf(buf, sizeof(buf), "%08lx", lb_id); |
||||
grpc::string lb_id_str(buf, kLbIdLength); |
||||
// The client may send requests with LB ID that has never been allocated
|
||||
// by this load reporter. Those IDs are tracked and will be skipped when
|
||||
// we generate a new ID.
|
||||
if (!load_data_store_.IsTrackedUnknownBalancerId(lb_id_str)) { |
||||
return lb_id_str; |
||||
} |
||||
} |
||||
} |
||||
|
||||
::grpc::lb::v1::LoadBalancingFeedback |
||||
LoadReporter::GenerateLoadBalancingFeedback() { |
||||
std::unique_lock<std::mutex> lock(feedback_mu_); |
||||
auto now = std::chrono::system_clock::now(); |
||||
// Discard records outside the window until there is only one record
|
||||
// outside the window, which is used as the base for difference.
|
||||
while (feedback_records_.size() > 1 && |
||||
!IsRecordInWindow(feedback_records_[1], now)) { |
||||
feedback_records_.pop_front(); |
||||
} |
||||
if (feedback_records_.size() < 2) { |
||||
return ::grpc::lb::v1::LoadBalancingFeedback::default_instance(); |
||||
} |
||||
// Find the longest range with valid ends.
|
||||
LoadBalancingFeedbackRecord* oldest = &feedback_records_[0]; |
||||
LoadBalancingFeedbackRecord* newest = |
||||
&feedback_records_[feedback_records_.size() - 1]; |
||||
while (newest > oldest && |
||||
(newest->cpu_limit == 0 || oldest->cpu_limit == 0)) { |
||||
// A zero limit means that the system info reading was failed, so these
|
||||
// records can't be used to calculate CPU utilization.
|
||||
if (newest->cpu_limit == 0) --newest; |
||||
if (oldest->cpu_limit == 0) ++oldest; |
||||
} |
||||
if (newest - oldest < 1 || oldest->end_time == newest->end_time || |
||||
newest->cpu_limit == oldest->cpu_limit) { |
||||
return ::grpc::lb::v1::LoadBalancingFeedback::default_instance(); |
||||
} |
||||
uint64_t rpcs = 0; |
||||
uint64_t errors = 0; |
||||
for (LoadBalancingFeedbackRecord* p = newest; p != oldest; --p) { |
||||
// Because these two numbers are counters, the oldest record shouldn't be
|
||||
// included.
|
||||
rpcs += p->rpcs; |
||||
errors += p->errors; |
||||
} |
||||
double cpu_usage = newest->cpu_usage - oldest->cpu_usage; |
||||
double cpu_limit = newest->cpu_limit - oldest->cpu_limit; |
||||
std::chrono::duration<double> duration_seconds = |
||||
newest->end_time - oldest->end_time; |
||||
lock.unlock(); |
||||
::grpc::lb::v1::LoadBalancingFeedback feedback; |
||||
feedback.set_server_utilization(static_cast<float>(cpu_usage / cpu_limit)); |
||||
feedback.set_calls_per_second( |
||||
static_cast<float>(rpcs / duration_seconds.count())); |
||||
feedback.set_errors_per_second( |
||||
static_cast<float>(errors / duration_seconds.count())); |
||||
return feedback; |
||||
} |
||||
|
||||
::google::protobuf::RepeatedPtrField<::grpc::lb::v1::Load> |
||||
LoadReporter::GenerateLoads(const grpc::string& hostname, |
||||
const grpc::string& lb_id) { |
||||
std::lock_guard<std::mutex> lock(store_mu_); |
||||
auto assigned_stores = load_data_store_.GetAssignedStores(hostname, lb_id); |
||||
GPR_ASSERT(assigned_stores != nullptr); |
||||
GPR_ASSERT(!assigned_stores->empty()); |
||||
::google::protobuf::RepeatedPtrField<::grpc::lb::v1::Load> loads; |
||||
for (PerBalancerStore* per_balancer_store : *assigned_stores) { |
||||
GPR_ASSERT(!per_balancer_store->IsSuspended()); |
||||
if (!per_balancer_store->load_record_map().empty()) { |
||||
for (const auto& p : per_balancer_store->load_record_map()) { |
||||
const auto& key = p.first; |
||||
const auto& value = p.second; |
||||
auto load = loads.Add(); |
||||
load->set_load_balance_tag(key.lb_tag()); |
||||
load->set_user_id(key.user_id()); |
||||
load->set_client_ip_address(key.GetClientIpBytes()); |
||||
load->set_num_calls_started(static_cast<int64_t>(value.start_count())); |
||||
load->set_num_calls_finished_without_error( |
||||
static_cast<int64_t>(value.ok_count())); |
||||
load->set_num_calls_finished_with_error( |
||||
static_cast<int64_t>(value.error_count())); |
||||
load->set_total_bytes_sent(static_cast<int64_t>(value.bytes_sent())); |
||||
load->set_total_bytes_received( |
||||
static_cast<int64_t>(value.bytes_recv())); |
||||
load->mutable_total_latency()->set_seconds( |
||||
static_cast<int64_t>(value.latency_ms() / 1000)); |
||||
load->mutable_total_latency()->set_nanos( |
||||
(static_cast<int32_t>(value.latency_ms()) % 1000) * 1000000); |
||||
for (const auto& p : value.call_metrics()) { |
||||
const grpc::string& metric_name = p.first; |
||||
const CallMetricValue& metric_value = p.second; |
||||
auto call_metric_data = load->add_metric_data(); |
||||
call_metric_data->set_metric_name(metric_name); |
||||
call_metric_data->set_num_calls_finished_with_metric( |
||||
metric_value.num_calls()); |
||||
call_metric_data->set_total_metric_value( |
||||
metric_value.total_metric_value()); |
||||
} |
||||
if (per_balancer_store->lb_id() != lb_id) { |
||||
// This per-balancer store is an orphan assigned to this receiving
|
||||
// balancer.
|
||||
AttachOrphanLoadId(load, *per_balancer_store); |
||||
} |
||||
} |
||||
per_balancer_store->ClearLoadRecordMap(); |
||||
} |
||||
if (per_balancer_store->IsNumCallsInProgressChangedSinceLastReport()) { |
||||
auto load = loads.Add(); |
||||
load->set_num_calls_in_progress( |
||||
per_balancer_store->GetNumCallsInProgressForReport()); |
||||
if (per_balancer_store->lb_id() != lb_id) { |
||||
// This per-balancer store is an orphan assigned to this receiving
|
||||
// balancer.
|
||||
AttachOrphanLoadId(load, *per_balancer_store); |
||||
} |
||||
} |
||||
} |
||||
return loads; |
||||
} |
||||
|
||||
void LoadReporter::AttachOrphanLoadId( |
||||
::grpc::lb::v1::Load* load, const PerBalancerStore& per_balancer_store) { |
||||
if (per_balancer_store.lb_id() == kInvalidLbId) { |
||||
load->set_load_key_unknown(true); |
||||
} else { |
||||
load->set_load_key_unknown(false); |
||||
load->mutable_orphaned_load_identifier()->set_load_key( |
||||
per_balancer_store.load_key()); |
||||
load->mutable_orphaned_load_identifier()->set_load_balancer_id( |
||||
per_balancer_store.lb_id()); |
||||
} |
||||
} |
||||
|
||||
void LoadReporter::AppendNewFeedbackRecord(uint64_t rpcs, uint64_t errors) { |
||||
CpuStatsProvider::CpuStatsSample cpu_stats; |
||||
if (cpu_stats_provider_ != nullptr) { |
||||
cpu_stats = cpu_stats_provider_->GetCpuStats(); |
||||
} else { |
||||
// This will make the load balancing feedback generation a no-op.
|
||||
cpu_stats = {0, 0}; |
||||
} |
||||
std::unique_lock<std::mutex> lock(feedback_mu_); |
||||
feedback_records_.emplace_back(std::chrono::system_clock::now(), rpcs, errors, |
||||
cpu_stats.first, cpu_stats.second); |
||||
} |
||||
|
||||
void LoadReporter::ReportStreamCreated(const grpc::string& hostname, |
||||
const grpc::string& lb_id, |
||||
const grpc::string& load_key) { |
||||
std::lock_guard<std::mutex> lock(store_mu_); |
||||
load_data_store_.ReportStreamCreated(hostname, lb_id, load_key); |
||||
gpr_log(GPR_INFO, |
||||
"[LR %p] Report stream created (host: %s, LB ID: %s, load key: %s).", |
||||
this, hostname.c_str(), lb_id.c_str(), load_key.c_str()); |
||||
} |
||||
|
||||
void LoadReporter::ReportStreamClosed(const grpc::string& hostname, |
||||
const grpc::string& lb_id) { |
||||
std::lock_guard<std::mutex> lock(store_mu_); |
||||
load_data_store_.ReportStreamClosed(hostname, lb_id); |
||||
gpr_log(GPR_INFO, "[LR %p] Report stream closed (host: %s, LB ID: %s).", this, |
||||
hostname.c_str(), lb_id.c_str()); |
||||
} |
||||
|
||||
void LoadReporter::ProcessViewDataCallStart( |
||||
const CensusViewProvider::ViewDataMap& view_data_map) { |
||||
auto it = view_data_map.find(kViewStartCount); |
||||
if (it != view_data_map.end()) { |
||||
// Note that the data type for any Sum view is double, whatever the data
|
||||
// type of the original measure.
|
||||
for (const auto& p : it->second.double_data()) { |
||||
const std::vector<grpc::string>& tag_values = p.first; |
||||
const uint64_t start_count = static_cast<uint64_t>(p.second); |
||||
const grpc::string& client_ip_and_token = tag_values[0]; |
||||
const grpc::string& host = tag_values[1]; |
||||
const grpc::string& user_id = tag_values[2]; |
||||
LoadRecordKey key(client_ip_and_token, user_id); |
||||
LoadRecordValue value = LoadRecordValue(start_count); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(store_mu_); |
||||
load_data_store_.MergeRow(host, key, value); |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
void LoadReporter::ProcessViewDataCallEnd( |
||||
const CensusViewProvider::ViewDataMap& view_data_map) { |
||||
uint64_t total_end_count = 0; |
||||
uint64_t total_error_count = 0; |
||||
auto it = view_data_map.find(kViewEndCount); |
||||
if (it != view_data_map.end()) { |
||||
// Note that the data type for any Sum view is double, whatever the data
|
||||
// type of the original measure.
|
||||
for (const auto& p : it->second.double_data()) { |
||||
const std::vector<grpc::string>& tag_values = p.first; |
||||
const uint64_t end_count = static_cast<uint64_t>(p.second); |
||||
const grpc::string& client_ip_and_token = tag_values[0]; |
||||
const grpc::string& host = tag_values[1]; |
||||
const grpc::string& user_id = tag_values[2]; |
||||
const grpc::string& status = tag_values[3]; |
||||
// This is due to a bug reported internally of Java server load reporting
|
||||
// implementation.
|
||||
// TODO(juanlishen): Check whether this situation happens in OSS C++.
|
||||
if (client_ip_and_token.size() == 0) { |
||||
gpr_log(GPR_DEBUG, |
||||
"Skipping processing Opencensus record with empty " |
||||
"client_ip_and_token tag."); |
||||
continue; |
||||
} |
||||
LoadRecordKey key(client_ip_and_token, user_id); |
||||
const uint64_t bytes_sent = |
||||
CensusViewProvider::GetRelatedViewDataRowDouble( |
||||
view_data_map, kViewEndBytesSent, sizeof(kViewEndBytesSent) - 1, |
||||
tag_values); |
||||
const uint64_t bytes_received = |
||||
CensusViewProvider::GetRelatedViewDataRowDouble( |
||||
view_data_map, kViewEndBytesReceived, |
||||
sizeof(kViewEndBytesReceived) - 1, tag_values); |
||||
const uint64_t latency_ms = |
||||
CensusViewProvider::GetRelatedViewDataRowDouble( |
||||
view_data_map, kViewEndLatencyMs, sizeof(kViewEndLatencyMs) - 1, |
||||
tag_values); |
||||
uint64_t ok_count = 0; |
||||
uint64_t error_count = 0; |
||||
total_end_count += end_count; |
||||
if (std::strcmp(status.c_str(), kCallStatusOk) == 0) { |
||||
ok_count = end_count; |
||||
} else { |
||||
error_count = end_count; |
||||
total_error_count += end_count; |
||||
} |
||||
LoadRecordValue value = LoadRecordValue( |
||||
0, ok_count, error_count, bytes_sent, bytes_received, latency_ms); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(store_mu_); |
||||
load_data_store_.MergeRow(host, key, value); |
||||
} |
||||
} |
||||
} |
||||
AppendNewFeedbackRecord(total_end_count, total_error_count); |
||||
} |
||||
|
||||
void LoadReporter::ProcessViewDataOtherCallMetrics( |
||||
const CensusViewProvider::ViewDataMap& view_data_map) { |
||||
auto it = view_data_map.find(kViewOtherCallMetricCount); |
||||
if (it != view_data_map.end()) { |
||||
for (const auto& p : it->second.int_data()) { |
||||
const std::vector<grpc::string>& tag_values = p.first; |
||||
const int64_t num_calls = p.second; |
||||
const grpc::string& client_ip_and_token = tag_values[0]; |
||||
const grpc::string& host = tag_values[1]; |
||||
const grpc::string& user_id = tag_values[2]; |
||||
const grpc::string& metric_name = tag_values[3]; |
||||
LoadRecordKey key(client_ip_and_token, user_id); |
||||
const double total_metric_value = |
||||
CensusViewProvider::GetRelatedViewDataRowDouble( |
||||
view_data_map, kViewOtherCallMetricValue, |
||||
sizeof(kViewOtherCallMetricValue) - 1, tag_values); |
||||
LoadRecordValue value = LoadRecordValue( |
||||
metric_name, static_cast<uint64_t>(num_calls), total_metric_value); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(store_mu_); |
||||
load_data_store_.MergeRow(host, key, value); |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
void LoadReporter::FetchAndSample() { |
||||
gpr_log(GPR_DEBUG, |
||||
"[LR %p] Starts fetching Census view data and sampling LB feedback " |
||||
"record.", |
||||
this); |
||||
CensusViewProvider::ViewDataMap view_data_map = |
||||
census_view_provider_->FetchViewData(); |
||||
ProcessViewDataCallStart(view_data_map); |
||||
ProcessViewDataCallEnd(view_data_map); |
||||
ProcessViewDataOtherCallMetrics(view_data_map); |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,225 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_SRC_CPP_SERVER_LOAD_REPORTER_LOAD_REPORTER_H |
||||
#define GRPC_SRC_CPP_SERVER_LOAD_REPORTER_LOAD_REPORTER_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <atomic> |
||||
#include <chrono> |
||||
#include <deque> |
||||
#include <vector> |
||||
|
||||
#include <grpc/support/log.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/load_data_store.h" |
||||
#include "src/proto/grpc/lb/v1/load_reporter.grpc.pb.h" |
||||
|
||||
#include "opencensus/stats/stats.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
// The interface to get the Census stats. Abstracted for mocking.
|
||||
class CensusViewProvider { |
||||
public: |
||||
// Maps from the view name to the view data.
|
||||
using ViewDataMap = |
||||
std::unordered_map<grpc::string, ::opencensus::stats::ViewData>; |
||||
// Maps from the view name to the view descriptor.
|
||||
using ViewDescriptorMap = |
||||
std::unordered_map<grpc::string, ::opencensus::stats::ViewDescriptor>; |
||||
|
||||
CensusViewProvider(); |
||||
virtual ~CensusViewProvider() = default; |
||||
|
||||
// Fetches the view data accumulated since last fetching, and returns it as a
|
||||
// map from the view name to the view data.
|
||||
virtual ViewDataMap FetchViewData() = 0; |
||||
|
||||
// A helper function that gets a row with the input tag values from the view
|
||||
// data. Only used when we know that row must exist because we have seen a row
|
||||
// with the same tag values in a related view data. Several ViewData's are
|
||||
// considered related if their views are based on the measures that are always
|
||||
// recorded at the same time.
|
||||
double static GetRelatedViewDataRowDouble( |
||||
const ViewDataMap& view_data_map, const char* view_name, |
||||
size_t view_name_len, const std::vector<grpc::string>& tag_values); |
||||
|
||||
protected: |
||||
const ViewDescriptorMap& view_descriptor_map() const { |
||||
return view_descriptor_map_; |
||||
} |
||||
|
||||
private: |
||||
ViewDescriptorMap view_descriptor_map_; |
||||
// Tag keys.
|
||||
::opencensus::stats::TagKey tag_key_token_; |
||||
::opencensus::stats::TagKey tag_key_host_; |
||||
::opencensus::stats::TagKey tag_key_user_id_; |
||||
::opencensus::stats::TagKey tag_key_status_; |
||||
::opencensus::stats::TagKey tag_key_metric_name_; |
||||
}; |
||||
|
||||
// The default implementation fetches the real stats from Census.
|
||||
class CensusViewProviderDefaultImpl : public CensusViewProvider { |
||||
public: |
||||
CensusViewProviderDefaultImpl(); |
||||
|
||||
ViewDataMap FetchViewData() override; |
||||
|
||||
private: |
||||
std::unordered_map<grpc::string, ::opencensus::stats::View> view_map_; |
||||
}; |
||||
|
||||
// The interface to get the CPU stats. Abstracted for mocking.
|
||||
class CpuStatsProvider { |
||||
public: |
||||
// The used and total amounts of CPU usage.
|
||||
using CpuStatsSample = std::pair<uint64_t, uint64_t>; |
||||
|
||||
virtual ~CpuStatsProvider() = default; |
||||
|
||||
// Gets the cumulative used CPU and total CPU resource.
|
||||
virtual CpuStatsSample GetCpuStats() = 0; |
||||
}; |
||||
|
||||
// The default implementation reads CPU jiffies from the system to calculate CPU
|
||||
// utilization.
|
||||
class CpuStatsProviderDefaultImpl : public CpuStatsProvider { |
||||
public: |
||||
CpuStatsSample GetCpuStats() override; |
||||
}; |
||||
|
||||
// Maintains all the load data and load reporting streams.
|
||||
class LoadReporter { |
||||
public: |
||||
// TODO(juanlishen): Allow config for providers from users.
|
||||
LoadReporter(uint32_t feedback_sample_window_seconds, |
||||
std::unique_ptr<CensusViewProvider> census_view_provider, |
||||
std::unique_ptr<CpuStatsProvider> cpu_stats_provider) |
||||
: feedback_sample_window_seconds_(feedback_sample_window_seconds), |
||||
census_view_provider_(std::move(census_view_provider)), |
||||
cpu_stats_provider_(std::move(cpu_stats_provider)) { |
||||
// Append the initial record so that the next real record can have a base.
|
||||
AppendNewFeedbackRecord(0, 0); |
||||
} |
||||
|
||||
// Fetches the latest data from Census and merge it into the data store.
|
||||
// Also adds a new sample to the LB feedback sliding window.
|
||||
// Thread-unsafe. (1). The access to the load data store and feedback records
|
||||
// has locking. (2). The access to the Census view provider and CPU stats
|
||||
// provider lacks locking, but we only access these two members in this method
|
||||
// (in testing, we also access them when setting up expectation). So the
|
||||
// invocations of this method must be serialized.
|
||||
void FetchAndSample(); |
||||
|
||||
// Generates a report for that host and balancer. The report contains
|
||||
// all the stats data accumulated between the last report (i.e., the last
|
||||
// consumption) and the last fetch from Census (i.e., the last production).
|
||||
// Thread-safe.
|
||||
::google::protobuf::RepeatedPtrField<::grpc::lb::v1::Load> GenerateLoads( |
||||
const grpc::string& hostname, const grpc::string& lb_id); |
||||
|
||||
// The feedback is calculated from the stats data recorded in the sliding
|
||||
// window. Outdated records are discarded.
|
||||
// Thread-safe.
|
||||
::grpc::lb::v1::LoadBalancingFeedback GenerateLoadBalancingFeedback(); |
||||
|
||||
// Wrapper around LoadDataStore::ReportStreamCreated.
|
||||
// Thread-safe.
|
||||
void ReportStreamCreated(const grpc::string& hostname, |
||||
const grpc::string& lb_id, |
||||
const grpc::string& load_key); |
||||
|
||||
// Wrapper around LoadDataStore::ReportStreamClosed.
|
||||
// Thread-safe.
|
||||
void ReportStreamClosed(const grpc::string& hostname, |
||||
const grpc::string& lb_id); |
||||
|
||||
// Generates a unique LB ID of length kLbIdLength. Returns an empty string
|
||||
// upon failure. Thread-safe.
|
||||
grpc::string GenerateLbId(); |
||||
|
||||
// Accessors only for testing.
|
||||
CensusViewProvider* census_view_provider() { |
||||
return census_view_provider_.get(); |
||||
} |
||||
CpuStatsProvider* cpu_stats_provider() { return cpu_stats_provider_.get(); } |
||||
|
||||
private: |
||||
struct LoadBalancingFeedbackRecord { |
||||
std::chrono::system_clock::time_point end_time; |
||||
uint64_t rpcs; |
||||
uint64_t errors; |
||||
uint64_t cpu_usage; |
||||
uint64_t cpu_limit; |
||||
|
||||
LoadBalancingFeedbackRecord( |
||||
const std::chrono::system_clock::time_point& end_time, uint64_t rpcs, |
||||
uint64_t errors, uint64_t cpu_usage, uint64_t cpu_limit) |
||||
: end_time(end_time), |
||||
rpcs(rpcs), |
||||
errors(errors), |
||||
cpu_usage(cpu_usage), |
||||
cpu_limit(cpu_limit) {} |
||||
}; |
||||
|
||||
// Finds the view data about starting call from the view_data_map and merges
|
||||
// the data to the load data store.
|
||||
void ProcessViewDataCallStart( |
||||
const CensusViewProvider::ViewDataMap& view_data_map); |
||||
// Finds the view data about ending call from the view_data_map and merges the
|
||||
// data to the load data store.
|
||||
void ProcessViewDataCallEnd( |
||||
const CensusViewProvider::ViewDataMap& view_data_map); |
||||
// Finds the view data about the customized call metrics from the
|
||||
// view_data_map and merges the data to the load data store.
|
||||
void ProcessViewDataOtherCallMetrics( |
||||
const CensusViewProvider::ViewDataMap& view_data_map); |
||||
|
||||
bool IsRecordInWindow(const LoadBalancingFeedbackRecord& record, |
||||
std::chrono::system_clock::time_point now) { |
||||
return record.end_time > now - feedback_sample_window_seconds_; |
||||
} |
||||
|
||||
void AppendNewFeedbackRecord(uint64_t rpcs, uint64_t errors); |
||||
|
||||
// Extracts an OrphanedLoadIdentifier from the per-balancer store and attaches
|
||||
// it to the load.
|
||||
void AttachOrphanLoadId(::grpc::lb::v1::Load* load, |
||||
const PerBalancerStore& per_balancer_store); |
||||
|
||||
std::atomic<int64_t> next_lb_id_{0}; |
||||
const std::chrono::seconds feedback_sample_window_seconds_; |
||||
std::mutex feedback_mu_; |
||||
std::deque<LoadBalancingFeedbackRecord> feedback_records_; |
||||
// TODO(juanlishen): Lock in finer grain. Locking the whole store may be
|
||||
// too expensive.
|
||||
std::mutex store_mu_; |
||||
LoadDataStore load_data_store_; |
||||
std::unique_ptr<CensusViewProvider> census_view_provider_; |
||||
std::unique_ptr<CpuStatsProvider> cpu_stats_provider_; |
||||
}; |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_SERVER_LOAD_REPORTER_LOAD_REPORTER_H
|
||||
@ -0,0 +1,92 @@ |
||||
|
||||
# gRPC iOS Network Transition Behaviors |
||||
Network connectivity on an iOS device may transition between cellular, WIFI, or |
||||
no network connectivity. This document describes how these network changes |
||||
should be handled by gRPC and current known issues. |
||||
|
||||
## Expected Network Transition Behaviors |
||||
The expected gRPC iOS channel and network transition behaviors are: |
||||
* Channel connection to a particular host is established at the time of |
||||
starting the first call to the channel and remains connected for future calls |
||||
to the same host. |
||||
* If the underlying connection to the remote host is broken, the channel is |
||||
disconnected and enters TRANSIENT\_FAILURE state. |
||||
* A channel is broken if the channel connection is no longer viable. This |
||||
happens when |
||||
* The network interface is no longer available, e.g. WiFi or cellular |
||||
interface is turned off or goes offline, airplane mode turned on, etc; |
||||
* The underlying TCP connection is no longer valid, e.g. WiFi connects to |
||||
another hotspot, cellular data switched from LTE to 4G, etc; |
||||
* A network interface more preferable by the OS is valid, e.g. WiFi gets |
||||
connected when the channel is already connected via cellular. |
||||
* A channel in TRANSIENT\_FAILURE state attempts reconnection on start of the |
||||
next call to the same host, but only after a certain backoff period (see |
||||
corresponding |
||||
[doc](https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md)). |
||||
During the backoff period, any call to the same host will wait until the |
||||
first of the following events occur: |
||||
* Connection succeeded; calls will be made using this channel; |
||||
* Conncetion failed; calls will be failed and return UNAVAILABLE status code; |
||||
* The call's deadline is reached; the call will fail and return |
||||
DEADLINE\_EXCEEDED status code. |
||||
The length of backoff period of a channel is reset whenever a connection |
||||
attempt is successful. |
||||
|
||||
## Implementations |
||||
### gRPC iOS with TCP Sockets |
||||
gRPC's default implementation is to use TCP sockets for networking. It turns |
||||
out that although Apple supports this type of usage, it is [not recommended by |
||||
Apple](https://developer.apple.com/library/archive/documentation/NetworkingInternetWeb/Conceptual/NetworkingOverview/SocketsAndStreams/SocketsAndStreams.html) |
||||
and has some issues described below. |
||||
|
||||
#### Issues with TCP Sockets |
||||
The TCP sockets on iOS is flawed in that it does not reflect the viability of |
||||
the channel connection. Particularly, we observed the following issues when |
||||
using TCP sockets: |
||||
* When a TCP socket connection is established on cellular data and WiFi |
||||
becomes available, the TCP socket neither return an error event nor continue |
||||
sending/receiving data on it, but still accepts write on it. |
||||
* A TCP socket does not report certain events that happen in the |
||||
background. When a TCP connection breaks in the background for the reason |
||||
like WiFi connects to another hotspot, the socket neither return an error nor |
||||
continue sending/receiving data on it, but still accepts write on it. |
||||
In both situations, the user will see the call hang for an extended period of |
||||
time before the TCP socket times out. |
||||
|
||||
#### gRPC iOS library's resolution to TCP socket issues |
||||
We introduced |
||||
[`ConnectivityMonitor`](https://developer.apple.com/library/archive/documentation/NetworkingInternetWeb/Conceptual/NetworkingOverview/SocketsAndStreams/SocketsAndStreams.html) |
||||
in gRPC iOS library v0.14.0 to alleviate these issues in TCP sockets, |
||||
which changes the network transition behaviors a bit. |
||||
|
||||
We classify network connectivity state of the device into three categories |
||||
based on flags obtained from `SCNetworkReachability` API: |
||||
|
||||
| Reachable | ConnectionRequired | IsWWAN | **Category** | |
||||
|:---------:|:------------------:|:------:|:------------:| |
||||
| 0 | X | X | None | |
||||
| X | 1 | X | None | |
||||
| 1 | 0 | 0 | WiFi | |
||||
| 1 | 0 | 1 | Cellular | |
||||
|
||||
Whenever there is a transition of network between two of these categories, all |
||||
previously existing channels are assumed to be broken and are actively |
||||
destroyed. If there is an unfinished call, the call should return with status |
||||
code `UNAVAILABLE`. |
||||
|
||||
`ConnectivityMonitor` is able to detect the scenario of the first issue above |
||||
and actively destroy the channels. However, the second issue is not resolvable. |
||||
To solve that issue the best solution is to switch to CFStream implementation |
||||
which eliminates all of them. |
||||
|
||||
### gRPC iOS with CFStream |
||||
gRPC iOS with CFStream implementation (introduced in v1.13.0) uses Apple's |
||||
networking API to make connections. It resolves the issues with TCP sockets |
||||
mentioned above. Users are recommended to use this implementation rather than |
||||
TCP socket implementation. The detailed behavior of streams in CFStream is not |
||||
documented by Apple, but our experiments show that it accords to the expected |
||||
behaviors. With CFStream implementation, an event is always received when the |
||||
underlying connection is no longer viable. For more detailed information and |
||||
usages of CFStream implementation, refer to the |
||||
[user guide](https://github.com/grpc/grpc/blob/master/src/objective-c/README-CFSTREAM.md). |
||||
|
||||
@ -0,0 +1,32 @@ |
||||
[](https://cocoapods.org/pods/gRPC) |
||||
# gRPC Objective-C with CFStream |
||||
|
||||
gRPC Objective-C library now provides the option to use Apple's CFStream API (rather than TCP |
||||
sockets) for networking. Using CFStream resolves a bunch of network connectivity transition issues |
||||
(see the [doc](https://github.com/grpc/grpc/blob/master/src/objective-c/NetworkTransitionBehavior.md) |
||||
for more information). |
||||
|
||||
CFStream integration is now in experimental state. You will need explicit opt-in to use it to get |
||||
the benefits of resolving the issues above. We expect to make CFStream the default networking |
||||
interface that gRPC uses when it is ready for production. |
||||
|
||||
## Usage |
||||
If you use gRPC following the instructions in |
||||
[README.md](https://github.com/grpc/grpc/blob/master/src/objective-c/README.md): |
||||
- Simply replace the |
||||
dependency on `gRPC-ProtoRPC` with `gRPC-ProtoRPC/CFStream`. The build system will take care of |
||||
everything else and switch networking to CFStream. |
||||
|
||||
If your project directly depends on podspecs other than `gRPC-ProtoRPC` (e.g. `gRPC` or |
||||
`gRPC-Core`): |
||||
|
||||
- Make your projects depend on subspecs corresponding to CFStream in each gRPC podspec. For |
||||
`gRPC-Core`, you will need to make sure that the completion queue you create is of type |
||||
`GRPC_CQ_NON_POLLING`. This is expected to be fixed soon so that you do not have to modify the |
||||
completion queue type. |
||||
|
||||
## Notes |
||||
|
||||
- Currently we do not support platforms other than iOS, although it is likely that this integration |
||||
can run on MacOS targets with Apple's compiler. |
||||
- Let us know if you meet any issue by filing issue and ping @muxi. |
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue