wu-sheng closed pull request #1001: Fix Auhentication token works incorrect
URL: https://github.com/apache/incubator-skywalking/pull/1001
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AuthenticationActivator.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AuthenticationDecorator.java
similarity index 73%
rename from
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AuthenticationActivator.java
rename to
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AuthenticationDecorator.java
index ddb61ac57..53c9ca898 100644
---
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AuthenticationActivator.java
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AuthenticationDecorator.java
@@ -18,28 +18,36 @@
package org.apache.skywalking.apm.agent.core.remote;
-import io.grpc.*;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ClientInterceptors;
+import io.grpc.ForwardingClientCall;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
import org.apache.skywalking.apm.agent.core.conf.Config;
import org.apache.skywalking.apm.util.StringUtil;
/**
* Active authentication header by Config.Agent.AUTHENTICATION
*
- * @author wu-sheng
+ * @author wu-sheng, zhang xin
*/
-public class AuthenticationActivator {
+public class AuthenticationDecorator implements ChannelDecorator {
private static final Metadata.Key<String> AUTH_HEAD_HEADER_NAME =
- Metadata.Key.of("Authentication",
Metadata.ASCII_STRING_MARSHALLER);
+ Metadata.Key.of("Authentication", Metadata.ASCII_STRING_MARSHALLER);
- public static Channel build(ManagedChannel originChannel) {
+ @Override
+ public Channel build(Channel channel) {
if (StringUtil.isEmpty(Config.Agent.AUTHENTICATION)) {
- return originChannel;
+ return channel;
}
- return ClientInterceptors.intercept(originChannel, new
ClientInterceptor() {
+ return ClientInterceptors.intercept(channel, new ClientInterceptor() {
@Override
public <REQ, RESP> ClientCall<REQ, RESP>
interceptCall(MethodDescriptor<REQ, RESP> method,
- CallOptions
options, Channel channel) {
+ CallOptions options, Channel channel) {
return new
ForwardingClientCall.SimpleForwardingClientCall<REQ,
RESP>(channel.newCall(method, options)) {
@Override
public void start(Listener<RESP> responseListener,
Metadata headers) {
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ChannelBuilder.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ChannelBuilder.java
new file mode 100644
index 000000000..f6b51c580
--- /dev/null
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ChannelBuilder.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ *
+ */
+
+package org.apache.skywalking.apm.agent.core.remote;
+
+import io.grpc.ManagedChannelBuilder;
+
+/**
+ * @author zhang xin
+ */
+public interface ChannelBuilder<B extends ManagedChannelBuilder> {
+ B build(B managedChannelBuilder) throws Exception;
+}
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ChannelDecorator.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ChannelDecorator.java
new file mode 100644
index 000000000..146d04e9f
--- /dev/null
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ChannelDecorator.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ *
+ */
+
+package org.apache.skywalking.apm.agent.core.remote;
+
+import io.grpc.Channel;
+
+/**
+ * @author zhang xin
+ */
+public interface ChannelDecorator {
+ Channel build(Channel channel);
+}
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannel.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannel.java
new file mode 100644
index 000000000..12756ccd5
--- /dev/null
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannel.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ *
+ */
+
+package org.apache.skywalking.apm.agent.core.remote;
+
+import io.grpc.Channel;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.netty.NettyChannelBuilder;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * @author zhangxin
+ */
+public class GRPCChannel {
+ /**
+ * origin channel
+ */
+ private final ManagedChannel originChannel;
+ private final Channel channelWithDecorators;
+
+ private GRPCChannel(String host, int port, List<ChannelBuilder>
channelBuilders,
+ List<ChannelDecorator> decorators) throws Exception {
+ ManagedChannelBuilder channelBuilder =
NettyChannelBuilder.forAddress(host, port);
+
+ for (ChannelBuilder builder : channelBuilders) {
+ channelBuilder = builder.build(channelBuilder);
+ }
+
+ this.originChannel = channelBuilder.build();
+
+ Channel channel = originChannel;
+ for (ChannelDecorator decorator : decorators) {
+ channel = decorator.build(channel);
+ }
+
+ channelWithDecorators = channel;
+ }
+
+ public static Builder newBuilder(String host, int port) {
+ return new Builder(host, port);
+ }
+
+ public Channel getChannel() {
+ return this.channelWithDecorators;
+ }
+
+ public boolean isTerminated() {
+ return originChannel.isTerminated();
+ }
+
+ public void shutdownNow() {
+ originChannel.shutdownNow();
+ }
+
+ public boolean isShutdown() {
+ return originChannel.isShutdown();
+ }
+
+ public static class Builder {
+ private final String host;
+ private final int port;
+ private final List<ChannelBuilder> channelBuilders;
+ private final List<ChannelDecorator> decorators;
+
+ private Builder(String host, int port) {
+ this.host = host;
+ this.port = port;
+ this.channelBuilders = new LinkedList<ChannelBuilder>();
+ this.decorators = new LinkedList<ChannelDecorator>();
+ }
+
+ public Builder addChannelDecorator(ChannelDecorator interceptor) {
+ this.decorators.add(interceptor);
+ return this;
+ }
+
+ public GRPCChannel build() throws Exception {
+ return new GRPCChannel(host, port, channelBuilders, decorators);
+ }
+
+ public Builder addManagedChannelBuilder(ChannelBuilder builder) {
+ channelBuilders.add(builder);
+ return this;
+ }
+ }
+}
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannelManager.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannelManager.java
index 212237d56..731fa3a20 100644
---
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannelManager.java
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannelManager.java
@@ -19,19 +19,8 @@
package org.apache.skywalking.apm.agent.core.remote;
import io.grpc.Channel;
-import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
-import io.grpc.internal.DnsNameResolverProvider;
-import io.grpc.netty.NettyChannelBuilder;
-import org.apache.skywalking.apm.agent.core.boot.BootService;
-import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;
-import org.apache.skywalking.apm.agent.core.conf.Config;
-import org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig;
-import org.apache.skywalking.apm.agent.core.logging.api.ILog;
-import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
-import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
-
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
@@ -39,15 +28,21 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import org.apache.skywalking.apm.agent.core.boot.BootService;
+import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;
+import org.apache.skywalking.apm.agent.core.conf.Config;
+import org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig;
+import org.apache.skywalking.apm.agent.core.logging.api.ILog;
+import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
+import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
/**
- * @author wusheng
+ * @author wusheng, zhang xin
*/
public class GRPCChannelManager implements BootService, Runnable {
private static final ILog logger =
LogManager.getLogger(GRPCChannelManager.class);
- private volatile ManagedChannel managedChannel = null;
- private volatile Channel publicChannelRef = null;
+ private volatile GRPCChannel managedChannel = null;
private volatile ScheduledFuture<?> connectCheckFuture;
private volatile boolean reconnect = true;
private Random random = new Random();
@@ -61,13 +56,13 @@ public void beforeBoot() throws Throwable {
@Override
public void boot() throws Throwable {
connectCheckFuture = Executors
- .newSingleThreadScheduledExecutor(new
DefaultNamedThreadFactory("GRPCChannelManager"))
- .scheduleAtFixedRate(new RunnableWithExceptionProtection(this,
new RunnableWithExceptionProtection.CallbackWhenException() {
- @Override
- public void handle(Throwable t) {
- logger.error("unexpected exception.", t);
- }
- }), 0, Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL,
TimeUnit.SECONDS);
+ .newSingleThreadScheduledExecutor(new
DefaultNamedThreadFactory("GRPCChannelManager"))
+ .scheduleAtFixedRate(new RunnableWithExceptionProtection(this, new
RunnableWithExceptionProtection.CallbackWhenException() {
+ @Override
+ public void handle(Throwable t) {
+ logger.error("unexpected exception.", t);
+ }
+ }), 0, Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL,
TimeUnit.SECONDS);
}
@Override
@@ -94,15 +89,13 @@ public void run() {
int index = Math.abs(random.nextInt()) %
RemoteDownstreamConfig.Collector.GRPC_SERVERS.size();
server =
RemoteDownstreamConfig.Collector.GRPC_SERVERS.get(index);
String[] ipAndPort = server.split(":");
- NettyChannelBuilder channelBuilder =
- new TLSChannelBuilder(
-
NettyChannelBuilder.forAddress(ipAndPort[0], Integer.parseInt(ipAndPort[1]))
- .nameResolverFactory(new
DnsNameResolverProvider())
- .maxInboundMessageSize(1024 * 1024
* 50)
- .usePlaintext(true)
- ).buildTLS();
- managedChannel = channelBuilder.build();
- publicChannelRef =
AuthenticationActivator.build(managedChannel);
+
+ managedChannel = GRPCChannel.newBuilder(ipAndPort[0],
Integer.parseInt(ipAndPort[1]))
+ .addManagedChannelBuilder(new StandardChannelBuilder())
+ .addManagedChannelBuilder(new TLSChannelBuilder())
+ .addChannelDecorator(new AuthenticationDecorator())
+ .build();
+
if (!managedChannel.isShutdown() &&
!managedChannel.isTerminated()) {
reconnect = false;
notify(GRPCChannelStatus.CONNECTED);
@@ -125,7 +118,7 @@ public void addChannelListener(GRPCChannelListener
listener) {
}
public Channel getChannel() {
- return managedChannel;
+ return managedChannel.getChannel();
}
/**
@@ -151,13 +144,13 @@ private void notify(GRPCChannelStatus status) {
private boolean isNetworkError(Throwable throwable) {
if (throwable instanceof StatusRuntimeException) {
- StatusRuntimeException statusRuntimeException =
(StatusRuntimeException) throwable;
+ StatusRuntimeException statusRuntimeException =
(StatusRuntimeException)throwable;
return statusEquals(statusRuntimeException.getStatus(),
- Status.UNAVAILABLE,
- Status.PERMISSION_DENIED,
- Status.UNAUTHENTICATED,
- Status.RESOURCE_EXHAUSTED,
- Status.UNKNOWN
+ Status.UNAVAILABLE,
+ Status.PERMISSION_DENIED,
+ Status.UNAUTHENTICATED,
+ Status.RESOURCE_EXHAUSTED,
+ Status.UNKNOWN
);
}
return false;
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/StandardChannelBuilder.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/StandardChannelBuilder.java
new file mode 100644
index 000000000..814d8d4d0
--- /dev/null
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/StandardChannelBuilder.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ *
+ */
+
+package org.apache.skywalking.apm.agent.core.remote;
+
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.internal.DnsNameResolverProvider;
+
+/**
+ * @author zhang xin
+ */
+public class StandardChannelBuilder implements ChannelBuilder {
+ private final static int MAX_INBOUND_MESSAGE_SIZE = 1024 * 1024 * 50;
+ private final static boolean USE_PLAIN_TEXT = true;
+
+ @Override public ManagedChannelBuilder build(ManagedChannelBuilder
managedChannelBuilder) throws Exception {
+ return managedChannelBuilder.nameResolverFactory(new
DnsNameResolverProvider())
+ .maxInboundMessageSize(MAX_INBOUND_MESSAGE_SIZE)
+ .usePlaintext(USE_PLAIN_TEXT);
+ }
+}
diff --git
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TLSChannelBuilder.java
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TLSChannelBuilder.java
index ce4569dbf..5dacdf4ce 100644
---
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TLSChannelBuilder.java
+++
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TLSChannelBuilder.java
@@ -22,42 +22,29 @@
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;
import io.netty.handler.ssl.SslContextBuilder;
+import java.io.File;
+import javax.net.ssl.SSLException;
import org.apache.skywalking.apm.agent.core.boot.AgentPackageNotFoundException;
import org.apache.skywalking.apm.agent.core.boot.AgentPackagePath;
import org.apache.skywalking.apm.agent.core.conf.Constants;
-import javax.net.ssl.SSLException;
-import java.io.File;
-
/**
* Detect the `/ca` folder in agent package, if `ca.crt` exists, start TLS (no
mutual auth).
*
* @author wusheng
*/
-public class TLSChannelBuilder {
+public class TLSChannelBuilder implements ChannelBuilder<NettyChannelBuilder> {
private static String CA_FILE_NAME = "ca" + Constants.PATH_SEPARATOR +
"ca.crt";
- private NettyChannelBuilder nettyChannelBuilder;
-
- public TLSChannelBuilder(NettyChannelBuilder nettyChannelBuilder) {
- this.nettyChannelBuilder = nettyChannelBuilder;
- }
-
- /**
- * Build a TLS supported channel is necessary.
- *
- * @return chanel builder
- * @throws AgentPackageNotFoundException
- * @throws SSLException
- */
- NettyChannelBuilder buildTLS() throws AgentPackageNotFoundException,
SSLException {
+ @Override public NettyChannelBuilder build(
+ NettyChannelBuilder managedChannelBuilder) throws
AgentPackageNotFoundException, SSLException {
File caFile = new File(AgentPackagePath.getPath(), CA_FILE_NAME);
if (caFile.exists() && caFile.isFile()) {
SslContextBuilder builder = GrpcSslContexts.forClient();
builder.trustManager(caFile);
- nettyChannelBuilder =
nettyChannelBuilder.negotiationType(NegotiationType.TLS)
- .sslContext(builder.build());
+ managedChannelBuilder =
managedChannelBuilder.negotiationType(NegotiationType.TLS)
+ .sslContext(builder.build());
}
- return nettyChannelBuilder;
+ return managedChannelBuilder;
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
