Skip to content

Commit df8cfe9

Browse files
ejona86ejona86
committedMay 29, 2024·
Create gcp-csm-observability
·
v1.71.0v1.65.0
1 parent 6dde844 commit df8cfe9

File tree

9 files changed

+1430
-1
lines changed

9 files changed

+1430
-1
lines changed
 

‎gcp-csm-observability/build.gradle

+31
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
plugins {
2+
id "java-library"
3+
4+
id "ru.vyarus.animalsniffer"
5+
}
6+
7+
description = "gRPC: GCP CSM Observability"
8+
9+
tasks.named("jar").configure {
10+
manifest {
11+
attributes('Automatic-Module-Name': 'io.grpc.gcp.csm.observability')
12+
}
13+
}
14+
15+
dependencies {
16+
implementation project(':grpc-api'),
17+
project(':grpc-core'),
18+
project(':grpc-opentelemetry'),
19+
project(':grpc-protobuf'),
20+
project(':grpc-xds'),
21+
libraries.guava.jre, // jre version pulled in via xds
22+
libraries.protobuf.java,
23+
libraries.opentelemetry.gcp.resources,
24+
libraries.opentelemetry.sdk.extension.autoconfigure // opentelemetry.gcp.resources uses compileOnly for this dep
25+
testImplementation project(":grpc-testing"),
26+
project(":grpc-inprocess"),
27+
libraries.opentelemetry.sdk.testing,
28+
libraries.assertj.core // opentelemetry.sdk.testing uses compileOnly for this dep
29+
30+
signature libraries.signature.java
31+
}

‎gcp-csm-observability/src/main/java/io/grpc/gcp/csm/observability/CsmObservability.java

+160
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
1+
/*
2+
* Copyright 2024 The gRPC Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.grpc.gcp.csm.observability;
18+
19+
import com.google.common.annotations.VisibleForTesting;
20+
import io.grpc.ExperimentalApi;
21+
import io.grpc.InternalConfigurator;
22+
import io.grpc.InternalConfiguratorRegistry;
23+
import io.grpc.ManagedChannelBuilder;
24+
import io.grpc.ServerBuilder;
25+
import io.grpc.opentelemetry.GrpcOpenTelemetry;
26+
import io.grpc.opentelemetry.InternalGrpcOpenTelemetry;
27+
import io.opentelemetry.api.OpenTelemetry;
28+
import java.io.Closeable;
29+
import java.util.Collection;
30+
import java.util.Collections;
31+
32+
/**
33+
* The entrypoint for GCP's CSM OpenTelemetry metrics functionality in gRPC.
34+
*
35+
* <p>CsmObservability uses {@link io.opentelemetry.api.OpenTelemetry} APIs for instrumentation.
36+
* When no SDK is explicitly added no telemetry data will be collected. See
37+
* {@code io.opentelemetry.sdk.OpenTelemetrySdk} for information on how to construct the SDK.
38+
*/
39+
@ExperimentalApi("TODO")
40+
public final class CsmObservability implements Closeable {
41+
private final GrpcOpenTelemetry delegate;
42+
private final MetadataExchanger exchanger;
43+
44+
public static Builder newBuilder() {
45+
return new Builder();
46+
}
47+
48+
private CsmObservability(Builder builder) {
49+
this.delegate = builder.delegate.build();
50+
this.exchanger = builder.exchanger;
51+
}
52+
53+
/**
54+
* Registers CsmObservability globally, applying its configuration to all subsequently created
55+
* gRPC channels and servers.
56+
*
57+
* <p>Note: Only one of CsmObservability and GrpcOpenTelemetry instance can be registered
58+
* globally. Any subsequent call to {@code registerGlobal()} will throw an {@code
59+
* IllegalStateException}.
60+
*/
61+
public void registerGlobal() {
62+
InternalConfiguratorRegistry.setConfigurators(Collections.singletonList(
63+
new InternalConfigurator() {
64+
@Override
65+
public void configureChannelBuilder(ManagedChannelBuilder<?> channelBuilder) {
66+
CsmObservability.this.configureChannelBuilder(channelBuilder);
67+
}
68+
69+
@Override
70+
public void configureServerBuilder(ServerBuilder<?> serverBuilder) {
71+
CsmObservability.this.configureServerBuilder(serverBuilder);
72+
}
73+
}));
74+
}
75+
76+
@VisibleForTesting
77+
void configureChannelBuilder(ManagedChannelBuilder<?> builder) {
78+
delegate.configureChannelBuilder(builder);
79+
}
80+
81+
@VisibleForTesting
82+
void configureServerBuilder(ServerBuilder<?> serverBuilder) {
83+
delegate.configureServerBuilder(serverBuilder);
84+
exchanger.configureServerBuilder(serverBuilder);
85+
}
86+
87+
@Override
88+
public void close() {}
89+
90+
/**
91+
* Builder for configuring {@link CsmObservability}.
92+
*/
93+
@ExperimentalApi("TODO")
94+
public static final class Builder {
95+
private final GrpcOpenTelemetry.Builder delegate = GrpcOpenTelemetry.newBuilder();
96+
private final MetadataExchanger exchanger;
97+
98+
private Builder() {
99+
this(new MetadataExchanger());
100+
}
101+
102+
@VisibleForTesting
103+
Builder(MetadataExchanger exchanger) {
104+
this.exchanger = exchanger;
105+
InternalGrpcOpenTelemetry.builderPlugin(delegate, exchanger);
106+
}
107+
108+
/**
109+
* Sets the {@link io.opentelemetry.api.OpenTelemetry} entrypoint to use. This can be used to
110+
* configure OpenTelemetry by returning the instance created by a
111+
* {@code io.opentelemetry.sdk.OpenTelemetrySdkBuilder}.
112+
*/
113+
public Builder sdk(OpenTelemetry sdk) {
114+
delegate.sdk(sdk);
115+
return this;
116+
}
117+
118+
/**
119+
* Adds optionalLabelKey to all the metrics that can provide value for the
120+
* optionalLabelKey.
121+
*/
122+
public Builder addOptionalLabel(String optionalLabelKey) {
123+
delegate.addOptionalLabel(optionalLabelKey);
124+
return this;
125+
}
126+
127+
/**
128+
* Enables the specified metrics for collection and export. By default, only a subset of
129+
* metrics are enabled.
130+
*/
131+
public Builder enableMetrics(Collection<String> enableMetrics) {
132+
delegate.enableMetrics(enableMetrics);
133+
return this;
134+
}
135+
136+
/**
137+
* Disables the specified metrics from being collected and exported.
138+
*/
139+
public Builder disableMetrics(Collection<String> disableMetrics) {
140+
delegate.disableMetrics(disableMetrics);
141+
return this;
142+
}
143+
144+
/**
145+
* Disable all metrics. If set to true all metrics must be explicitly enabled.
146+
*/
147+
public Builder disableAllMetrics() {
148+
delegate.disableAllMetrics();
149+
return this;
150+
}
151+
152+
/**
153+
* Returns a new {@link CsmObservability} built with the configuration of this {@link
154+
* Builder}.
155+
*/
156+
public CsmObservability build() {
157+
return new CsmObservability(this);
158+
}
159+
}
160+
}

‎gcp-csm-observability/src/main/java/io/grpc/gcp/csm/observability/MetadataExchanger.java

+364
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,364 @@
1+
/*
2+
* Copyright 2024 The gRPC Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.grpc.gcp.csm.observability;
18+
19+
import com.google.common.annotations.VisibleForTesting;
20+
import com.google.common.base.Preconditions;
21+
import com.google.common.io.BaseEncoding;
22+
import com.google.protobuf.Struct;
23+
import com.google.protobuf.Value;
24+
import io.grpc.CallOptions;
25+
import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
26+
import io.grpc.Metadata;
27+
import io.grpc.ServerBuilder;
28+
import io.grpc.ServerCall;
29+
import io.grpc.ServerCallHandler;
30+
import io.grpc.ServerInterceptor;
31+
import io.grpc.Status;
32+
import io.grpc.internal.JsonParser;
33+
import io.grpc.internal.JsonUtil;
34+
import io.grpc.opentelemetry.InternalOpenTelemetryPlugin;
35+
import io.grpc.protobuf.ProtoUtils;
36+
import io.grpc.xds.ClusterImplLoadBalancerProvider;
37+
import io.grpc.xds.InternalGrpcBootstrapperImpl;
38+
import io.opentelemetry.api.common.AttributeKey;
39+
import io.opentelemetry.api.common.Attributes;
40+
import io.opentelemetry.api.common.AttributesBuilder;
41+
import io.opentelemetry.contrib.gcp.resource.GCPResourceProvider;
42+
import java.net.URI;
43+
import java.util.Map;
44+
import java.util.function.Consumer;
45+
import java.util.logging.Level;
46+
import java.util.logging.Logger;
47+
48+
/**
49+
* OpenTelemetryPlugin implementing metadata-based workload property exchange for both client and
50+
* server. Is responsible for determining the metadata, communicating the metadata, and adding local
51+
* and remote details to metrics.
52+
*/
53+
final class MetadataExchanger implements InternalOpenTelemetryPlugin {
54+
private static final Logger logger = Logger.getLogger(MetadataExchanger.class.getName());
55+
56+
private static final AttributeKey<String> CLOUD_PLATFORM =
57+
AttributeKey.stringKey("cloud.platform");
58+
private static final AttributeKey<String> K8S_NAMESPACE_NAME =
59+
AttributeKey.stringKey("k8s.namespace.name");
60+
private static final AttributeKey<String> K8S_CLUSTER_NAME =
61+
AttributeKey.stringKey("k8s.cluster.name");
62+
private static final AttributeKey<String> CLOUD_AVAILABILITY_ZONE =
63+
AttributeKey.stringKey("cloud.availability_zone");
64+
private static final AttributeKey<String> CLOUD_REGION =
65+
AttributeKey.stringKey("cloud.region");
66+
private static final AttributeKey<String> CLOUD_ACCOUNT_ID =
67+
AttributeKey.stringKey("cloud.account.id");
68+
69+
private static final Metadata.Key<String> SEND_KEY =
70+
Metadata.Key.of("x-envoy-peer-metadata", Metadata.ASCII_STRING_MARSHALLER);
71+
private static final Metadata.Key<Struct> RECV_KEY =
72+
Metadata.Key.of("x-envoy-peer-metadata", new BinaryToAsciiMarshaller<>(
73+
ProtoUtils.metadataMarshaller(Struct.getDefaultInstance())));
74+
75+
private static final String EXCHANGE_TYPE = "type";
76+
private static final String EXCHANGE_CANONICAL_SERVICE = "canonical_service";
77+
private static final String EXCHANGE_PROJECT_ID = "project_id";
78+
private static final String EXCHANGE_LOCATION = "location";
79+
private static final String EXCHANGE_CLUSTER_NAME = "cluster_name";
80+
private static final String EXCHANGE_NAMESPACE_NAME = "namespace_name";
81+
private static final String EXCHANGE_WORKLOAD_NAME = "workload_name";
82+
private static final String TYPE_GKE = "gcp_kubernetes_engine";
83+
private static final String TYPE_GCE = "gcp_compute_engine";
84+
85+
private final String localMetadata;
86+
private final Attributes localAttributes;
87+
88+
public MetadataExchanger() {
89+
this(
90+
new GCPResourceProvider().getAttributes(),
91+
System::getenv,
92+
InternalGrpcBootstrapperImpl::getJsonContent);
93+
}
94+
95+
MetadataExchanger(Attributes platformAttributes, Lookup env, Supplier<String> xdsBootstrap) {
96+
String type = platformAttributes.get(CLOUD_PLATFORM);
97+
String canonicalService = env.get("CSM_CANONICAL_SERVICE_NAME");
98+
Struct.Builder struct = Struct.newBuilder();
99+
put(struct, EXCHANGE_TYPE, type);
100+
put(struct, EXCHANGE_CANONICAL_SERVICE, canonicalService);
101+
if (TYPE_GKE.equals(type)) {
102+
String location = platformAttributes.get(CLOUD_AVAILABILITY_ZONE);
103+
if (location == null) {
104+
location = platformAttributes.get(CLOUD_REGION);
105+
}
106+
put(struct, EXCHANGE_WORKLOAD_NAME, env.get("CSM_WORKLOAD_NAME"));
107+
put(struct, EXCHANGE_NAMESPACE_NAME, platformAttributes.get(K8S_NAMESPACE_NAME));
108+
put(struct, EXCHANGE_CLUSTER_NAME, platformAttributes.get(K8S_CLUSTER_NAME));
109+
put(struct, EXCHANGE_LOCATION, location);
110+
put(struct, EXCHANGE_PROJECT_ID, platformAttributes.get(CLOUD_ACCOUNT_ID));
111+
} else if (TYPE_GCE.equals(type)) {
112+
String location = platformAttributes.get(CLOUD_AVAILABILITY_ZONE);
113+
if (location == null) {
114+
location = platformAttributes.get(CLOUD_REGION);
115+
}
116+
put(struct, EXCHANGE_WORKLOAD_NAME, env.get("CSM_WORKLOAD_NAME"));
117+
put(struct, EXCHANGE_LOCATION, location);
118+
put(struct, EXCHANGE_PROJECT_ID, platformAttributes.get(CLOUD_ACCOUNT_ID));
119+
}
120+
localMetadata = BaseEncoding.base64().encode(struct.build().toByteArray());
121+
122+
localAttributes = Attributes.builder()
123+
.put("csm.mesh_id", nullIsUnknown(getMeshId(xdsBootstrap)))
124+
.put("csm.workload_canonical_service", nullIsUnknown(canonicalService))
125+
.build();
126+
}
127+
128+
private static String nullIsUnknown(String value) {
129+
return value == null ? "unknown" : value;
130+
}
131+
132+
private static void put(Struct.Builder struct, String key, String value) {
133+
value = nullIsUnknown(value);
134+
struct.putFields(key, Value.newBuilder().setStringValue(value).build());
135+
}
136+
137+
private static void put(AttributesBuilder attributes, String key, Value value) {
138+
attributes.put(key, nullIsUnknown(fromValue(value)));
139+
}
140+
141+
private static String fromValue(Value value) {
142+
if (value == null) {
143+
return null;
144+
}
145+
if (value.getKindCase() != Value.KindCase.STRING_VALUE) {
146+
return null;
147+
}
148+
return value.getStringValue();
149+
}
150+
151+
@VisibleForTesting
152+
static String getMeshId(Supplier<String> xdsBootstrap) {
153+
try {
154+
@SuppressWarnings("unchecked")
155+
Map<String, ?> rawBootstrap = (Map<String, ?>) JsonParser.parse(xdsBootstrap.get());
156+
Map<String, ?> node = JsonUtil.getObject(rawBootstrap, "node");
157+
String id = JsonUtil.getString(node, "id");
158+
Preconditions.checkNotNull(id, "id");
159+
String[] parts = id.split("/", 6);
160+
if (!(parts.length == 6
161+
&& parts[0].equals("projects")
162+
&& parts[2].equals("networks")
163+
&& parts[3].startsWith("mesh:")
164+
&& parts[4].equals("nodes"))) {
165+
throw new Exception("node id didn't match mesh format: " + id);
166+
}
167+
return parts[3].substring("mesh:".length());
168+
} catch (Exception e) {
169+
logger.log(Level.INFO, "Failed to determine mesh ID for CSM", e);
170+
return null;
171+
}
172+
}
173+
174+
private void addLabels(AttributesBuilder to, Struct struct) {
175+
to.putAll(localAttributes);
176+
Map<String, Value> remote = struct.getFieldsMap();
177+
Value typeValue = remote.get(EXCHANGE_TYPE);
178+
String type = fromValue(typeValue);
179+
put(to, "csm.remote_workload_type", typeValue);
180+
put(to, "csm.remote_workload_canonical_service", remote.get(EXCHANGE_CANONICAL_SERVICE));
181+
if (TYPE_GKE.equals(type)) {
182+
put(to, "csm.remote_workload_project_id", remote.get(EXCHANGE_PROJECT_ID));
183+
put(to, "csm.remote_workload_location", remote.get(EXCHANGE_LOCATION));
184+
put(to, "csm.remote_workload_cluster_name", remote.get(EXCHANGE_CLUSTER_NAME));
185+
put(to, "csm.remote_workload_namespace_name", remote.get(EXCHANGE_NAMESPACE_NAME));
186+
put(to, "csm.remote_workload_name", remote.get(EXCHANGE_WORKLOAD_NAME));
187+
} else if (TYPE_GCE.equals(type)) {
188+
put(to, "csm.remote_workload_project_id", remote.get(EXCHANGE_PROJECT_ID));
189+
put(to, "csm.remote_workload_location", remote.get(EXCHANGE_LOCATION));
190+
put(to, "csm.remote_workload_name", remote.get(EXCHANGE_WORKLOAD_NAME));
191+
}
192+
}
193+
194+
@Override
195+
public boolean enablePluginForChannel(String target) {
196+
URI uri;
197+
try {
198+
uri = new URI(target);
199+
} catch (Exception ex) {
200+
return false;
201+
}
202+
String authority = uri.getAuthority();
203+
return "xds".equals(uri.getScheme())
204+
&& (authority == null || "traffic-director-global.xds.googleapis.com".equals(authority));
205+
}
206+
207+
@Override
208+
public ClientCallPlugin newClientCallPlugin() {
209+
return new ClientCallState();
210+
}
211+
212+
public void configureServerBuilder(ServerBuilder<?> serverBuilder) {
213+
serverBuilder.intercept(new ServerCallInterceptor());
214+
}
215+
216+
@Override
217+
public ServerStreamPlugin newServerStreamPlugin(Metadata inboundMetadata) {
218+
return new ServerStreamState(inboundMetadata.get(RECV_KEY));
219+
}
220+
221+
final class ClientCallState implements ClientCallPlugin {
222+
private volatile Value serviceName;
223+
private volatile Value serviceNamespace;
224+
225+
@Override
226+
public ClientStreamPlugin newClientStreamPlugin() {
227+
return new ClientStreamState();
228+
}
229+
230+
@Override
231+
public CallOptions filterCallOptions(CallOptions options) {
232+
Consumer<Map<String, Struct>> existingConsumer =
233+
options.getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER);
234+
return options.withOption(
235+
ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER,
236+
(Map<String, Struct> clusterMetadata) -> {
237+
metadataConsumer(clusterMetadata);
238+
existingConsumer.accept(clusterMetadata);
239+
});
240+
}
241+
242+
private void metadataConsumer(Map<String, Struct> clusterMetadata) {
243+
Struct struct = clusterMetadata.get("com.google.csm.telemetry_labels");
244+
if (struct == null) {
245+
struct = Struct.getDefaultInstance();
246+
}
247+
serviceName = struct.getFieldsMap().get("service_name");
248+
serviceNamespace = struct.getFieldsMap().get("service_namespace");
249+
}
250+
251+
@Override
252+
public void addMetadata(Metadata toMetadata) {
253+
toMetadata.put(SEND_KEY, localMetadata);
254+
}
255+
256+
class ClientStreamState implements ClientStreamPlugin {
257+
private Struct receivedExchange;
258+
259+
@Override
260+
public void inboundHeaders(Metadata headers) {
261+
setExchange(headers);
262+
}
263+
264+
@Override
265+
public void inboundTrailers(Metadata trailers) {
266+
if (receivedExchange != null) {
267+
return; // Received headers
268+
}
269+
setExchange(trailers);
270+
}
271+
272+
private void setExchange(Metadata metadata) {
273+
Struct received = metadata.get(RECV_KEY);
274+
if (received == null) {
275+
receivedExchange = Struct.getDefaultInstance();
276+
} else {
277+
receivedExchange = received;
278+
}
279+
}
280+
281+
@Override
282+
public void addLabels(AttributesBuilder to) {
283+
put(to, "csm.service_name", serviceName);
284+
put(to, "csm.service_namespace", serviceNamespace);
285+
Struct exchange = receivedExchange;
286+
if (exchange == null) {
287+
exchange = Struct.getDefaultInstance();
288+
}
289+
MetadataExchanger.this.addLabels(to, exchange);
290+
}
291+
}
292+
}
293+
294+
final class ServerCallInterceptor implements ServerInterceptor {
295+
@Override
296+
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
297+
ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
298+
if (!headers.containsKey(RECV_KEY)) {
299+
return next.startCall(call, headers);
300+
} else {
301+
return next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
302+
private boolean headersSent;
303+
304+
@Override
305+
public void sendHeaders(Metadata headers) {
306+
headersSent = true;
307+
headers.put(SEND_KEY, localMetadata);
308+
super.sendHeaders(headers);
309+
}
310+
311+
@Override
312+
public void close(Status status, Metadata trailers) {
313+
if (!headersSent) {
314+
trailers.put(SEND_KEY, localMetadata);
315+
}
316+
super.close(status, trailers);
317+
}
318+
}, headers);
319+
}
320+
}
321+
}
322+
323+
final class ServerStreamState implements ServerStreamPlugin {
324+
private final Struct receivedExchange;
325+
326+
ServerStreamState(Struct exchange) {
327+
if (exchange == null) {
328+
exchange = Struct.getDefaultInstance();
329+
}
330+
receivedExchange = exchange;
331+
}
332+
333+
@Override
334+
public void addLabels(AttributesBuilder to) {
335+
MetadataExchanger.this.addLabels(to, receivedExchange);
336+
}
337+
}
338+
339+
interface Lookup {
340+
String get(String name);
341+
}
342+
343+
interface Supplier<T> {
344+
T get() throws Exception;
345+
}
346+
347+
static final class BinaryToAsciiMarshaller<T> implements Metadata.AsciiMarshaller<T> {
348+
private final Metadata.BinaryMarshaller<T> delegate;
349+
350+
public BinaryToAsciiMarshaller(Metadata.BinaryMarshaller<T> delegate) {
351+
this.delegate = Preconditions.checkNotNull(delegate, "delegate");
352+
}
353+
354+
@Override
355+
public T parseAsciiString(String serialized) {
356+
return delegate.parseBytes(BaseEncoding.base64().decode(serialized));
357+
}
358+
359+
@Override
360+
public String toAsciiString(T value) {
361+
return BaseEncoding.base64().encode(delegate.toBytes(value));
362+
}
363+
}
364+
}

‎gcp-csm-observability/src/test/java/io/grpc/gcp/csm/observability/CsmObservabilityTest.java

+636
Large diffs are not rendered by default.

‎gcp-csm-observability/src/test/java/io/grpc/gcp/csm/observability/MetadataExchangerTest.java

+201
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,201 @@
1+
/*
2+
* Copyright 2024 The gRPC Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.grpc.gcp.csm.observability;
18+
19+
import static com.google.common.truth.Truth.assertThat;
20+
import static io.opentelemetry.api.common.AttributeKey.stringKey;
21+
22+
import com.google.common.collect.ImmutableMap;
23+
import com.google.common.io.BaseEncoding;
24+
import com.google.protobuf.Struct;
25+
import com.google.protobuf.Value;
26+
import io.grpc.Metadata;
27+
import io.opentelemetry.api.common.Attributes;
28+
import io.opentelemetry.api.common.AttributesBuilder;
29+
import org.junit.Test;
30+
import org.junit.runner.RunWith;
31+
import org.junit.runners.JUnit4;
32+
33+
/** Tests for {@link MetadataExchanger}. */
34+
@RunWith(JUnit4.class)
35+
public final class MetadataExchangerTest {
36+
@Test
37+
public void getMeshId_findsMeshId() {
38+
assertThat(MetadataExchanger.getMeshId(() ->
39+
"{\"node\":{\"id\":\"projects/12/networks/mesh:mine/nodes/uu-id\"}}"))
40+
.isEqualTo("mine");
41+
assertThat(MetadataExchanger.getMeshId(() ->
42+
"{\"node\":{\"id\":\"projects/1234567890/networks/mesh:mine/nodes/uu-id\", "
43+
+ "\"unknown\": \"\"}, \"unknown\": \"\"}"))
44+
.isEqualTo("mine");
45+
}
46+
47+
@Test
48+
public void getMeshId_returnsNullOnBadMeshId() {
49+
assertThat(MetadataExchanger.getMeshId(
50+
() -> "[\"node\"]"))
51+
.isNull();
52+
assertThat(MetadataExchanger.getMeshId(
53+
() -> "{\"node\":[\"id\"]}}"))
54+
.isNull();
55+
assertThat(MetadataExchanger.getMeshId(
56+
() -> "{\"node\":{\"id\":[\"projects/12/networks/mesh:mine/nodes/uu-id\"]}}"))
57+
.isNull();
58+
59+
assertThat(MetadataExchanger.getMeshId(
60+
() -> "{\"NODE\":{\"id\":\"projects/12/networks/mesh:mine/nodes/uu-id\"}}"))
61+
.isNull();
62+
assertThat(MetadataExchanger.getMeshId(
63+
() -> "{\"node\":{\"ID\":\"projects/12/networks/mesh:mine/nodes/uu-id\"}}"))
64+
.isNull();
65+
assertThat(MetadataExchanger.getMeshId(
66+
() -> "{\"node\":{\"id\":\"projects/12/networks/mesh:mine\"}}"))
67+
.isNull();
68+
assertThat(MetadataExchanger.getMeshId(
69+
() -> "{\"node\":{\"id\":\"PROJECTS/12/networks/mesh:mine/nodes/uu-id\"}}"))
70+
.isNull();
71+
assertThat(MetadataExchanger.getMeshId(
72+
() -> "{\"node\":{\"id\":\"projects/12/NETWORKS/mesh:mine/nodes/uu-id\"}}"))
73+
.isNull();
74+
assertThat(MetadataExchanger.getMeshId(
75+
() -> "{\"node\":{\"id\":\"projects/12/networks/MESH:mine/nodes/uu-id\"}}"))
76+
.isNull();
77+
assertThat(MetadataExchanger.getMeshId(
78+
() -> "{\"node\":{\"id\":\"projects/12/networks/mesh:mine/NODES/uu-id\"}}"))
79+
.isNull();
80+
}
81+
82+
@Test
83+
public void enablePluginForChannel_matches() {
84+
MetadataExchanger exchanger =
85+
new MetadataExchanger(Attributes.builder().build(), (name) -> null, () -> "");
86+
assertThat(exchanger.enablePluginForChannel("xds:///testing")).isTrue();
87+
assertThat(exchanger.enablePluginForChannel("xds:/testing")).isTrue();
88+
assertThat(exchanger.enablePluginForChannel(
89+
"xds://traffic-director-global.xds.googleapis.com/testing:123")).isTrue();
90+
}
91+
92+
@Test
93+
public void enablePluginForChannel_doesNotMatch() {
94+
MetadataExchanger exchanger =
95+
new MetadataExchanger(Attributes.builder().build(), (name) -> null, () -> "");
96+
assertThat(exchanger.enablePluginForChannel("dns:///localhost")).isFalse();
97+
assertThat(exchanger.enablePluginForChannel("xds:///[]")).isFalse();
98+
assertThat(exchanger.enablePluginForChannel("xds://my-xds-server/testing")).isFalse();
99+
}
100+
101+
@Test
102+
public void addLabels_receivedWrongType() {
103+
MetadataExchanger exchanger =
104+
new MetadataExchanger(Attributes.builder().build(), (name) -> null, () -> "");
105+
Metadata metadata = new Metadata();
106+
metadata.put(Metadata.Key.of("x-envoy-peer-metadata", Metadata.ASCII_STRING_MARSHALLER),
107+
BaseEncoding.base64().encode(Struct.newBuilder()
108+
.putFields("type", Value.newBuilder().setNumberValue(1).build())
109+
.build()
110+
.toByteArray()));
111+
AttributesBuilder builder = Attributes.builder();
112+
exchanger.newServerStreamPlugin(metadata).addLabels(builder);
113+
114+
assertThat(builder.build()).isEqualTo(Attributes.builder()
115+
.put(stringKey("csm.mesh_id"), "unknown")
116+
.put(stringKey("csm.workload_canonical_service"), "unknown")
117+
.put(stringKey("csm.remote_workload_type"), "unknown")
118+
.put(stringKey("csm.remote_workload_canonical_service"), "unknown")
119+
.build());
120+
}
121+
122+
@Test
123+
public void addLabelsFromExchange_unknownGcpType() {
124+
MetadataExchanger exchanger =
125+
new MetadataExchanger(Attributes.builder().build(), (name) -> null, () -> "");
126+
Metadata metadata = new Metadata();
127+
metadata.put(Metadata.Key.of("x-envoy-peer-metadata", Metadata.ASCII_STRING_MARSHALLER),
128+
BaseEncoding.base64().encode(Struct.newBuilder()
129+
.putFields("type", Value.newBuilder().setStringValue("gcp_surprise").build())
130+
.putFields("canonical_service", Value.newBuilder().setStringValue("myservice1").build())
131+
.build()
132+
.toByteArray()));
133+
AttributesBuilder builder = Attributes.builder();
134+
exchanger.newServerStreamPlugin(metadata).addLabels(builder);
135+
136+
assertThat(builder.build()).isEqualTo(Attributes.builder()
137+
.put(stringKey("csm.mesh_id"), "unknown")
138+
.put(stringKey("csm.workload_canonical_service"), "unknown")
139+
.put(stringKey("csm.remote_workload_type"), "gcp_surprise")
140+
.put(stringKey("csm.remote_workload_canonical_service"), "myservice1")
141+
.build());
142+
}
143+
144+
@Test
145+
public void addMetadata_k8s() throws Exception {
146+
MetadataExchanger exchanger = new MetadataExchanger(
147+
Attributes.builder()
148+
.put(stringKey("cloud.platform"), "gcp_kubernetes_engine")
149+
.put(stringKey("k8s.namespace.name"), "mynamespace1")
150+
.put(stringKey("k8s.cluster.name"), "mycluster1")
151+
.put(stringKey("cloud.availability_zone"), "myzone1")
152+
.put(stringKey("cloud.account.id"), "0001")
153+
.build(),
154+
ImmutableMap.of(
155+
"CSM_CANONICAL_SERVICE_NAME", "myservice1",
156+
"CSM_WORKLOAD_NAME", "myworkload1")::get,
157+
() -> "");
158+
Metadata metadata = new Metadata();
159+
exchanger.newClientCallPlugin().addMetadata(metadata);
160+
161+
Struct peer = Struct.parseFrom(BaseEncoding.base64().decode(metadata.get(
162+
Metadata.Key.of("x-envoy-peer-metadata", Metadata.ASCII_STRING_MARSHALLER))));
163+
assertThat(peer).isEqualTo(
164+
Struct.newBuilder()
165+
.putFields("type", Value.newBuilder().setStringValue("gcp_kubernetes_engine").build())
166+
.putFields("canonical_service", Value.newBuilder().setStringValue("myservice1").build())
167+
.putFields("workload_name", Value.newBuilder().setStringValue("myworkload1").build())
168+
.putFields("namespace_name", Value.newBuilder().setStringValue("mynamespace1").build())
169+
.putFields("cluster_name", Value.newBuilder().setStringValue("mycluster1").build())
170+
.putFields("location", Value.newBuilder().setStringValue("myzone1").build())
171+
.putFields("project_id", Value.newBuilder().setStringValue("0001").build())
172+
.build());
173+
}
174+
175+
@Test
176+
public void addMetadata_gce() throws Exception {
177+
MetadataExchanger exchanger = new MetadataExchanger(
178+
Attributes.builder()
179+
.put(stringKey("cloud.platform"), "gcp_compute_engine")
180+
.put(stringKey("cloud.availability_zone"), "myzone1")
181+
.put(stringKey("cloud.account.id"), "0001")
182+
.build(),
183+
ImmutableMap.of(
184+
"CSM_CANONICAL_SERVICE_NAME", "myservice1",
185+
"CSM_WORKLOAD_NAME", "myworkload1")::get,
186+
() -> "");
187+
Metadata metadata = new Metadata();
188+
exchanger.newClientCallPlugin().addMetadata(metadata);
189+
190+
Struct peer = Struct.parseFrom(BaseEncoding.base64().decode(metadata.get(
191+
Metadata.Key.of("x-envoy-peer-metadata", Metadata.ASCII_STRING_MARSHALLER))));
192+
assertThat(peer).isEqualTo(
193+
Struct.newBuilder()
194+
.putFields("type", Value.newBuilder().setStringValue("gcp_compute_engine").build())
195+
.putFields("canonical_service", Value.newBuilder().setStringValue("myservice1").build())
196+
.putFields("workload_name", Value.newBuilder().setStringValue("myworkload1").build())
197+
.putFields("location", Value.newBuilder().setStringValue("myzone1").build())
198+
.putFields("project_id", Value.newBuilder().setStringValue("0001").build())
199+
.build());
200+
}
201+
}

‎gradle/libs.versions.toml

+2
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,8 @@ opencensus-exporter-trace-stackdriver = { module = "io.opencensus:opencensus-exp
7777
opencensus-impl = { module = "io.opencensus:opencensus-impl", version.ref = "opencensus" }
7878
opencensus-proto = "io.opencensus:opencensus-proto:0.2.0"
7979
opentelemetry-api = "io.opentelemetry:opentelemetry-api:1.36.0"
80+
opentelemetry-gcp-resources = "io.opentelemetry.contrib:opentelemetry-gcp-resources:1.34.0-alpha"
81+
opentelemetry-sdk-extension-autoconfigure = "io.opentelemetry:opentelemetry-sdk-extension-autoconfigure:1.36.0"
8082
opentelemetry-sdk-testing = "io.opentelemetry:opentelemetry-sdk-testing:1.36.0"
8183
perfmark-api = "io.perfmark:perfmark-api:0.26.0"
8284
protobuf-java = { module = "com.google.protobuf:protobuf-java", version.ref = "protobuf" }

‎settings.gradle

+2
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ include ":grpc-xds"
6868
include ":grpc-bom"
6969
include ":grpc-rls"
7070
include ":grpc-authz"
71+
include ":grpc-gcp-csm-observability"
7172
include ":grpc-gcp-observability"
7273
include ":grpc-gcp-observability:interop"
7374
include ":grpc-istio-interop-testing"
@@ -102,6 +103,7 @@ project(':grpc-xds').projectDir = "$rootDir/xds" as File
102103
project(':grpc-bom').projectDir = "$rootDir/bom" as File
103104
project(':grpc-rls').projectDir = "$rootDir/rls" as File
104105
project(':grpc-authz').projectDir = "$rootDir/authz" as File
106+
project(':grpc-gcp-csm-observability').projectDir = "$rootDir/gcp-csm-observability" as File
105107
project(':grpc-gcp-observability').projectDir = "$rootDir/gcp-observability" as File
106108
project(':grpc-gcp-observability:interop').projectDir = "$rootDir/gcp-observability/interop" as File
107109
project(':grpc-istio-interop-testing').projectDir = "$rootDir/istio-interop-testing" as File

‎testing/src/main/java/io/grpc/internal/testing/FakeNameResolverProvider.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ protected boolean isAvailable() {
5252

5353
@Override
5454
protected int priority() {
55-
return 5; // Default
55+
return 10; // High priority
5656
}
5757

5858
@Override

‎xds/src/main/java/io/grpc/xds/InternalGrpcBootstrapperImpl.java

+33
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Copyright 2024 The gRPC Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.grpc.xds;
18+
19+
import io.grpc.Internal;
20+
import io.grpc.xds.client.XdsInitializationException;
21+
import java.io.IOException;
22+
23+
/**
24+
* Internal accessors for GrpcBootstrapperImpl.
25+
*/
26+
@Internal
27+
public final class InternalGrpcBootstrapperImpl {
28+
private InternalGrpcBootstrapperImpl() {} // prevent instantiation
29+
30+
public static String getJsonContent() throws XdsInitializationException, IOException {
31+
return new GrpcBootstrapperImpl().getJsonContent();
32+
}
33+
}

0 commit comments

Comments
 (0)
Please sign in to comment.