From f53d8a53d74b44e7907bd1fb4055117577548c54 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 4 Jun 2021 16:40:40 +0800 Subject: [PATCH] Introduce cert signer for xDS, opt xDS Impl (#7934) --- dubbo-dependencies-bom/pom.xml | 17 ++ dubbo-registry/dubbo-registry-xds/pom.xml | 46 ++++ .../registry/xds/XdsCertificateSigner.java | 52 ++++ .../org/apache/dubbo/registry/xds/XdsEnv.java | 21 ++ .../registry/xds/XdsServiceDiscovery.java | 22 +- .../xds/XdsServiceDiscoveryFactory.java | 12 +- .../istio/IstioCitadelCertificateSigner.java | 249 ++++++++++++++++++ .../registry/xds/istio/IstioConstant.java | 53 ++++ .../dubbo/registry/xds/istio/IstioEnv.java | 121 +++++++++ .../registry/xds/util/PilotExchanger.java | 43 ++- .../dubbo/registry/xds/util/XdsChannel.java | 32 ++- .../xds/util/protocol/AbstractProtocol.java | 156 +++++------ .../xds/util/protocol/impl/EdsProtocol.java | 25 +- .../xds/util/protocol/impl/LdsProtocol.java | 31 +-- .../xds/util/protocol/impl/RdsProtocol.java | 25 +- .../src/main/proto/ca.proto | 57 ++++ ...he.dubbo.registry.xds.XdsCertificateSigner | 1 + pom.xml | 2 + 18 files changed, 783 insertions(+), 182 deletions(-) create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/proto/ca.proto create mode 100644 dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml index 5cf588e5b2..9ab091283c 100644 --- a/dubbo-dependencies-bom/pom.xml +++ b/dubbo-dependencies-bom/pom.xml @@ -161,6 +161,7 @@ 1.5.19 0.9.0 + 1.68 2.0.1 5.2.0 2.8.5 @@ -344,6 +345,22 @@ RoaringBitmap ${roaringbitmap_version} + + org.bouncycastle + bcprov-jdk15on + ${bouncycastle-bcprov_version} + + + org.bouncycastle + bcpkix-jdk15on + ${bouncycastle-bcprov_version} + + + org.bouncycastle + bcprov-ext-jdk15on + ${bouncycastle-bcprov_version} + + javax.annotation diff --git a/dubbo-registry/dubbo-registry-xds/pom.xml b/dubbo-registry/dubbo-registry-xds/pom.xml index effda5aa93..7e1ca4ae8e 100644 --- a/dubbo-registry/dubbo-registry-xds/pom.xml +++ b/dubbo-registry/dubbo-registry-xds/pom.xml @@ -53,6 +53,11 @@ grpc-stub + + io.grpc + grpc-netty-shaded + + io.envoyproxy.controlplane api @@ -63,6 +68,47 @@ protobuf-java-util + + org.bouncycastle + bcprov-jdk15on + + + org.bouncycastle + bcpkix-jdk15on + + + org.bouncycastle + bcprov-ext-jdk15on + + + + + + kr.motd.maven + os-maven-plugin + 1.6.2 + + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + + com.google.protobuf:protoc:3.5.1-1:exe:${os.detected.classifier} + grpc-java + io.grpc:protoc-gen-grpc-java:1.14.0:exe:${os.detected.classifier} + + + + + compile + compile-custom + + + + + + \ No newline at end of file diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java new file mode 100644 index 0000000000..d3b9047269 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java @@ -0,0 +1,52 @@ +/* + * 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.dubbo.registry.xds; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.extension.Adaptive; +import org.apache.dubbo.common.extension.SPI; + +@SPI +public interface XdsCertificateSigner { + + @Adaptive(value = "Signer") + CertPair request(URL url); + + class CertPair { + private final String privateKey; + private final String publicKey; + private final long expireTime; + + public CertPair(String privateKey, String publicKey, long expireTime) { + this.privateKey = privateKey; + this.publicKey = publicKey; + this.expireTime = expireTime; + } + + public String getPrivateKey() { + return privateKey; + } + + public String getPublicKey() { + return publicKey; + } + + public boolean isExpire() { + return System.currentTimeMillis() < expireTime; + } + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java new file mode 100644 index 0000000000..e0fa40c2e7 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java @@ -0,0 +1,21 @@ +/* + * 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.dubbo.registry.xds; + +public interface XdsEnv { + String getCluster(); +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java index 0651e3ede4..a2bded3db1 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java @@ -17,6 +17,8 @@ package org.apache.dubbo.registry.xds; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.registry.client.DefaultServiceInstance; import org.apache.dubbo.registry.client.SelfHostMetaServiceDiscovery; import org.apache.dubbo.registry.client.ServiceInstance; @@ -32,11 +34,15 @@ import java.util.Set; public class XdsServiceDiscovery extends SelfHostMetaServiceDiscovery { private PilotExchanger exchanger; - private URL registryURL; + private static final Logger logger = LoggerFactory.getLogger(XdsServiceDiscovery.class); @Override public void doInitialize(URL registryURL) throws Exception { - exchanger = PilotExchanger.initialize(registryURL); + try { + exchanger = PilotExchanger.initialize(registryURL); + } catch (Throwable t) { + logger.error(t); + } } @Override @@ -67,10 +73,14 @@ public class XdsServiceDiscovery extends SelfHostMetaServiceDiscovery { private List changedToInstances(String serviceName, Collection endpoints) { List instances = new LinkedList<>(); endpoints.forEach(endpoint -> { - DefaultServiceInstance serviceInstance = new DefaultServiceInstance(serviceName, endpoint.getAddress(), endpoint.getPortValue()); - // fill metadata by SelfHostMetaServiceDiscovery, will be fetched by RPC request - fillServiceInstance(serviceInstance); - instances.add(serviceInstance); + try { + DefaultServiceInstance serviceInstance = new DefaultServiceInstance(serviceName, endpoint.getAddress(), endpoint.getPortValue()); + // fill metadata by SelfHostMetaServiceDiscovery, will be fetched by RPC request + fillServiceInstance(serviceInstance); + instances.add(serviceInstance); + } catch (Throwable t) { + logger.error("Error occurred when parsing endpoints. Endpoints List:" + endpoints,t); + } }); instances.sort(Comparator.comparingInt(ServiceInstance::hashCode)); return instances; diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java index ef77ccbbc9..19d15cf21c 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java @@ -17,12 +17,22 @@ package org.apache.dubbo.registry.xds; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.registry.client.AbstractServiceDiscoveryFactory; import org.apache.dubbo.registry.client.ServiceDiscovery; public class XdsServiceDiscoveryFactory extends AbstractServiceDiscoveryFactory { + private static final Logger logger = LoggerFactory.getLogger(XdsServiceDiscoveryFactory.class); + @Override protected ServiceDiscovery createDiscovery(URL registryURL) { - return new XdsServiceDiscovery(); + XdsServiceDiscovery xdsServiceDiscovery = new XdsServiceDiscovery(); + try { + xdsServiceDiscovery.doInitialize(registryURL); + } catch (Exception e) { + logger.error("Error occurred when initialize xDS service discovery impl.", e); + } + return xdsServiceDiscovery; } } diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java new file mode 100644 index 0000000000..ceaf858a54 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java @@ -0,0 +1,249 @@ +/* + * 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.dubbo.registry.xds.istio; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.XdsCertificateSigner; +import org.apache.dubbo.rpc.RpcException; + +import io.grpc.ManagedChannel; +import io.grpc.Metadata; +import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; +import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; +import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory; +import io.grpc.stub.MetadataUtils; +import io.grpc.stub.StreamObserver; +import istio.v1.auth.Ca; +import istio.v1.auth.IstioCertificateServiceGrpc; +import org.bouncycastle.asn1.pkcs.PKCSObjectIdentifiers; +import org.bouncycastle.asn1.x500.X500Name; +import org.bouncycastle.asn1.x509.Extension; +import org.bouncycastle.asn1.x509.ExtensionsGenerator; +import org.bouncycastle.asn1.x509.GeneralName; +import org.bouncycastle.asn1.x509.GeneralNames; +import org.bouncycastle.openssl.jcajce.JcaPEMWriter; +import org.bouncycastle.operator.ContentSigner; +import org.bouncycastle.operator.OperatorCreationException; +import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder; +import org.bouncycastle.pkcs.PKCS10CertificationRequest; +import org.bouncycastle.pkcs.jcajce.JcaPKCS10CertificationRequestBuilder; +import org.bouncycastle.util.io.pem.PemObject; + +import java.io.IOException; +import java.io.StringWriter; +import java.security.InvalidAlgorithmParameterException; +import java.security.KeyPair; +import java.security.KeyPairGenerator; +import java.security.NoSuchAlgorithmException; +import java.security.PrivateKey; +import java.security.PublicKey; +import java.security.SecureRandom; +import java.security.spec.ECGenParameterSpec; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; + +public class IstioCitadelCertificateSigner implements XdsCertificateSigner { + + private static final Logger logger = LoggerFactory.getLogger(IstioCitadelCertificateSigner.class); + + private final IstioEnv istioEnv; + + private CertPair certPairCache; + + public IstioCitadelCertificateSigner() { + istioEnv = new IstioEnv(); + } + + @Override + public CertPair request(URL url) { + if (certPairCache != null && !certPairCache.isExpire()) { + return certPairCache; + } + synchronized (this) { + if (certPairCache == null || certPairCache.isExpire()) { + try { + certPairCache = createCert(url); + } catch (IOException e) { + logger.error("Generate Cert from Istio failed.", e); + throw new RpcException("Generate Cert from Istio failed.", e); + } + } + } + return certPairCache; + } + + public CertPair createCert(URL url) throws IOException { + PublicKey publicKey = null; + PrivateKey privateKey = null; + ContentSigner signer = null; + + if (istioEnv.isECCFirst()) { + try { + ECGenParameterSpec ecSpec = new ECGenParameterSpec("secp256r1"); + KeyPairGenerator g = KeyPairGenerator.getInstance("EC"); + g.initialize(ecSpec, new SecureRandom()); + KeyPair keypair = g.generateKeyPair(); + publicKey = keypair.getPublic(); + privateKey = keypair.getPrivate(); + + signer = new JcaContentSignerBuilder("SHA256withECDSA").build(keypair.getPrivate()); + + } catch (NoSuchAlgorithmException | InvalidAlgorithmParameterException | OperatorCreationException e) { + logger.error("Generate Key with secp256r1 algorithm failed. Please check if your system support. " + + "Will attempt to generate with RSA2048.", e); + } + } + + if (publicKey == null) { + try { + KeyPairGenerator kpGenerator = KeyPairGenerator.getInstance("RSA"); + kpGenerator.initialize(istioEnv.getRasKeySize()); + KeyPair keypair = kpGenerator.generateKeyPair(); + publicKey = keypair.getPublic(); + privateKey = keypair.getPrivate(); + signer = new JcaContentSignerBuilder("SHA256WithRSA").build(keypair.getPrivate()); + + } catch (NoSuchAlgorithmException | OperatorCreationException e) { + logger.error("Generate Key with SHA256WithRSA algorithm failed. Please check if your system support.", e); + throw new RpcException(e); + } + } + + String csr = generateCsr(publicKey, signer); + + ManagedChannel channel = NettyChannelBuilder.forTarget(istioEnv.getCaAddr()) + .sslContext(GrpcSslContexts.forClient() + .trustManager(InsecureTrustManagerFactory.INSTANCE) + .build()) + .build(); + + Metadata header = new Metadata(); + Metadata.Key key = Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER); + header.put(key, "Bearer " + istioEnv.getServiceAccount()); + + key = Metadata.Key.of("ClusterID", Metadata.ASCII_STRING_MARSHALLER); + header.put(key, istioEnv.getIstioMetaClusterId()); + + IstioCertificateServiceGrpc.IstioCertificateServiceStub stub = IstioCertificateServiceGrpc.newStub(channel); + + stub = MetadataUtils.attachHeaders(stub, header); + + CountDownLatch countDownLatch = new CountDownLatch(1); + StringBuffer publicKeyBuilder = new StringBuffer(); + AtomicBoolean failed = new AtomicBoolean(false); + + stub.createCertificate( + generateRequest(csr), + generateResponseObserver(countDownLatch, publicKeyBuilder, failed)); + + long expireTime = System.currentTimeMillis() + (long) (istioEnv.getSecretTTL() * istioEnv.getSecretGracePeriodRatio()); + + try { + countDownLatch.await(); + } catch (InterruptedException e) { + throw new RpcException("Generate Cert Failed. Wait for cert failed.", e); + } + + if (failed.get()) { + throw new RpcException("Generate Cert Failed. Send csr request failed. Please check log above."); + } + + String privateKeyPem = generatePrivatePemKey(privateKey); + CertPair certPair = new CertPair(privateKeyPem, publicKeyBuilder.toString(), expireTime); + + channel.shutdown(); + return certPair; + } + + private Ca.IstioCertificateRequest generateRequest(String csr) { + return Ca.IstioCertificateRequest + .newBuilder() + .setCsr(csr) + .setValidityDuration(istioEnv.getSecretTTL()) + .build(); + } + + private StreamObserver generateResponseObserver(CountDownLatch countDownLatch, StringBuffer publicKeyBuilder, AtomicBoolean failed) { + return new StreamObserver() { + @Override + public void onNext(Ca.IstioCertificateResponse istioCertificateResponse) { + for (int i = 0; i < istioCertificateResponse.getCertChainCount(); i++) { + publicKeyBuilder.append(istioCertificateResponse.getCertChainBytes(i).toStringUtf8()); + } + if (logger.isDebugEnabled()) { + logger.debug("Receive Cert chain from Istio Citadel. \n" + publicKeyBuilder); + } + countDownLatch.countDown(); + } + + @Override + public void onError(Throwable throwable) { + failed.set(true); + logger.error("Receive error message from Istio Citadel grpc stub.", throwable); + countDownLatch.countDown(); + } + + @Override + public void onCompleted() { + countDownLatch.countDown(); + } + }; + } + + private String generatePrivatePemKey(PrivateKey privateKey) throws IOException { + String key = generatePemKey("RSA PRIVATE KEY", privateKey.getEncoded()); + if (logger.isDebugEnabled()) { + logger.debug("Generated Private Key. \n" + key); + } + return key; + } + + private String generatePemKey(String type, byte[] content) throws IOException { + PemObject pemObject = new PemObject(type, content); + StringWriter str = new StringWriter(); + JcaPEMWriter jcaPEMWriter = new JcaPEMWriter(str); + jcaPEMWriter.writeObject(pemObject); + jcaPEMWriter.close(); + str.close(); + return str.toString(); + } + + private String generateCsr(PublicKey publicKey, ContentSigner signer) throws IOException { + GeneralNames subjectAltNames = new GeneralNames( + new GeneralName[]{ + new GeneralName(6, istioEnv.getCsrHost()) + }); + + ExtensionsGenerator extGen = new ExtensionsGenerator(); + extGen.addExtension(Extension.subjectAlternativeName, true, subjectAltNames); + + PKCS10CertificationRequest request = + new JcaPKCS10CertificationRequestBuilder( + new X500Name("O=" + istioEnv.getTrustDomain()), publicKey) + .addAttribute(PKCSObjectIdentifiers.pkcs_9_at_extensionRequest, extGen.generate()) + .build(signer); + + String csr = generatePemKey("CERTIFICATE REQUEST", request.getEncoded()); + + if (logger.isDebugEnabled()) { + logger.debug("CSR Request to Istio Citadel. \n" + csr); + } + return csr; + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java new file mode 100644 index 0000000000..8e71eea48f --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java @@ -0,0 +1,53 @@ +/* + * 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.dubbo.registry.xds.istio; + +public class IstioConstant { + public final static String CA_ADDR_KEY = "CA_ADDR"; + + public final static String DEFAULT_CA_ADDR = "istiod.istio-system.svc:15012"; + + public final static String TRUST_DOMAIN_KEY = "TRUST_DOMAIN"; + + public final static String DEFAULT_TRUST_DOMAIN = "cluster.local"; + + public final static String WORKLOAD_NAMESPACE_KEY = "WORKLOAD_NAMESPACE"; + + public final static String DEFAULT_WORKLOAD_NAMESPACE = "default"; + + public final static String KUBERNETES_SA_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token"; + + public final static String RSA_KEY_SIZE_KEY = "RSA_KEY_SIZE"; + + public final static String DEFAULT_RSA_KEY_SIZE = "2048"; + + public final static String ECC_SIG_ALG_KEY = "ECC_SIGNATURE_ALGORITHM"; + + public final static String DEFAULT_ECC_SIG_ALG = "ECDSA"; + + public final static String SECRET_TTL_KEY = "SECRET_TTL"; + + public final static String DEFAULT_SECRET_TTL = "86400"; //24 * 60 * 60 + + public final static String SECRET_GRACE_PERIOD_RATIO_KEY = "SECRET_GRACE_PERIOD_RATIO"; + + public final static String DEFAULT_SECRET_GRACE_PERIOD_RATIO = "0.5"; + + public final static String ISTIO_META_CLUSTER_ID_KEY = "ISTIO_META_CLUSTER_ID"; + + public final static String DEFAULT_ISTIO_META_CLUSTER_ID = "kubernetes"; +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java new file mode 100644 index 0000000000..3aa4de5988 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java @@ -0,0 +1,121 @@ +/* + * 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.dubbo.registry.xds.istio; + +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.XdsEnv; + +import org.apache.commons.io.FileUtils; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Optional; + +public class IstioEnv implements XdsEnv { + private static final Logger logger = LoggerFactory.getLogger(IstioEnv.class); + + private String caAddr; + + private String serviceAccount = null; + + private String csrHost; + + private String trustDomain; + + private String workloadNameSpace; + + private int rasKeySize; + + private String eccSigAlg; + + private int secretTTL; + + private float secretGracePeriodRatio; + + private String istioMetaClusterId; + + public IstioEnv() { + File saFile = new File(IstioConstant.KUBERNETES_SA_PATH); + if (saFile.canRead()) { + try { + serviceAccount = FileUtils.readFileToString(saFile, StandardCharsets.UTF_8); + trustDomain = Optional.ofNullable(System.getenv(IstioConstant.TRUST_DOMAIN_KEY)).orElse(IstioConstant.DEFAULT_TRUST_DOMAIN); + workloadNameSpace = Optional.ofNullable(System.getenv(IstioConstant.WORKLOAD_NAMESPACE_KEY)).orElse(IstioConstant.DEFAULT_WORKLOAD_NAMESPACE); + csrHost = "spiffe://" + trustDomain + "/ns/" + workloadNameSpace + "/sa/" + serviceAccount; + caAddr = Optional.ofNullable(System.getenv(IstioConstant.CA_ADDR_KEY)).orElse(IstioConstant.DEFAULT_CA_ADDR); + rasKeySize = Integer.parseInt(Optional.ofNullable(System.getenv(IstioConstant.RSA_KEY_SIZE_KEY)).orElse(IstioConstant.DEFAULT_RSA_KEY_SIZE)); + eccSigAlg = Optional.ofNullable(System.getenv(IstioConstant.ECC_SIG_ALG_KEY)).orElse(IstioConstant.DEFAULT_ECC_SIG_ALG); + secretTTL = Integer.parseInt(Optional.ofNullable(System.getenv(IstioConstant.SECRET_TTL_KEY)).orElse(IstioConstant.DEFAULT_SECRET_TTL)); + secretGracePeriodRatio = Float.parseFloat(Optional.ofNullable(System.getenv(IstioConstant.SECRET_GRACE_PERIOD_RATIO_KEY)).orElse(IstioConstant.DEFAULT_SECRET_GRACE_PERIOD_RATIO)); + istioMetaClusterId = Optional.ofNullable(System.getenv(IstioConstant.ISTIO_META_CLUSTER_ID_KEY)).orElse(IstioConstant.DEFAULT_ISTIO_META_CLUSTER_ID); + } catch (IOException e) { + logger.error("Unable to read token file.", e); + } + } + if (serviceAccount == null) { + throw new UnsupportedOperationException("Unable to found kubernetes service account token file. " + + "Please check if work in Kubernetes and mount service account token file correctly."); + } + } + + public String getCaAddr() { + return caAddr; + } + + public String getServiceAccount() { + return serviceAccount; + } + + public String getCsrHost() { + return csrHost; + } + + public String getTrustDomain() { + return trustDomain; + } + + public String getWorkloadNameSpace() { + return workloadNameSpace; + } + + @Override + public String getCluster() { + return null; + } + + public int getRasKeySize() { + return rasKeySize; + } + + public boolean isECCFirst() { + return "ECDSA".equals(eccSigAlg); + } + + public int getSecretTTL() { + return secretTTL; + } + + public float getSecretGracePeriodRatio() { + return secretGracePeriodRatio; + } + + public String getIstioMetaClusterId() { + return istioMetaClusterId; + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java index 3d49bc1a9f..6d63985376 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java @@ -31,6 +31,7 @@ import java.util.Collections; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; public class PilotExchanger { @@ -45,7 +46,7 @@ public class PilotExchanger { private RouteResult routeResult; - private final long observeRouteRequest; + private final AtomicLong observeRouteRequest = new AtomicLong(-1); private final Map domainObserveRequest = new ConcurrentHashMap<>(); @@ -53,15 +54,39 @@ public class PilotExchanger { private PilotExchanger(URL url) { xdsChannel = new XdsChannel(url); - LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build()); - this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build()); - this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build()); + int pollingPoolSize = url.getParameter("pollingPoolSize", 10); + int pollingTimeout = url.getParameter("pollingTimeout", 10); + LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout); + this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout); + this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout); this.listenerResult = ldsProtocol.getListeners(); this.routeResult = rdsProtocol.getResource(listenerResult.getRouteConfigNames()); // Observer RDS update - this.observeRouteRequest = rdsProtocol.observeResource(listenerResult.getRouteConfigNames(), (newResult) -> { + if (CollectionUtils.isNotEmpty(listenerResult.getRouteConfigNames())) { + this.observeRouteRequest.set(createRouteObserve()); + } + + // Observe LDS updated + ldsProtocol.observeListeners((newListener) -> { + // update local cache + if (!newListener.equals(listenerResult)) { + this.listenerResult = newListener; + // update RDS observation + synchronized (observeRouteRequest) { + if (observeRouteRequest.get() == -1) { + this.observeRouteRequest.set(createRouteObserve()); + } else { + rdsProtocol.updateObserve(observeRouteRequest.get(), newListener.getRouteConfigNames()); + } + } + } + }); + } + + private long createRouteObserve() { + return rdsProtocol.observeResource(listenerResult.getRouteConfigNames(), (newResult) -> { // check if observed domain update ( will update endpoint observation ) domainObserveConsumer.forEach((domain, consumer) -> { Set newRoute = newResult.searchDomain(domain); @@ -81,14 +106,6 @@ public class PilotExchanger { // update local cache routeResult = newResult; }); - - // Observe LDS updated - ldsProtocol.observeListeners((newListener) -> { - // update local cache - this.listenerResult = newListener; - // update RDS observation - rdsProtocol.updateObserve(observeRouteRequest, newListener.getRouteConfigNames()); - }); } public static PilotExchanger initialize(URL url) { diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java index 5336378909..088729e397 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java @@ -17,6 +17,10 @@ package org.apache.dubbo.registry.xds.util; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.extension.ExtensionLoader; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.XdsCertificateSigner; import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc; import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryRequest; @@ -24,16 +28,36 @@ import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; +import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; +import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; +import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext; +import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory; import io.grpc.stub.StreamObserver; +import javax.net.ssl.SSLException; +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; + public class XdsChannel { private final ManagedChannel channel; + private static final Logger logger = LoggerFactory.getLogger(XdsChannel.class); protected XdsChannel(URL url) { - channel = ManagedChannelBuilder.forAddress(url.getHost(), url.getPort()) - .usePlaintext() - .build(); + ManagedChannel channel1 = null; + try { + XdsCertificateSigner signer = ExtensionLoader.getExtensionLoader(XdsCertificateSigner.class).getExtension(url.getParameter("Signer","istio")); + XdsCertificateSigner.CertPair certPair = signer.request(url); + SslContext context = GrpcSslContexts.forClient() + .trustManager(InsecureTrustManagerFactory.INSTANCE) + .keyManager(new ByteArrayInputStream(certPair.getPublicKey().getBytes(StandardCharsets.UTF_8)), new ByteArrayInputStream(certPair.getPrivateKey().getBytes(StandardCharsets.UTF_8))) + .build(); + channel1 = NettyChannelBuilder.forAddress(url.getHost(), url.getPort()) + .sslContext(context) + .build(); + } catch (SSLException e) { + logger.error("Error occurred when creating gRPC channel to control panel.", e); + } + channel = channel1; } public StreamObserver observeDeltaDiscoveryRequest(StreamObserver observer) { diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java index 13ac5be347..b689a7aac0 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java @@ -18,25 +18,28 @@ package org.apache.dubbo.registry.xds.util.protocol; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.NamedThreadFactory; import org.apache.dubbo.registry.xds.util.XdsChannel; import io.envoyproxy.envoy.config.core.v3.Node; -import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryRequest; -import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; import io.grpc.stub.StreamObserver; -import java.util.HashSet; +import java.util.Collections; import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; -public abstract class AbstractProtocol> implements XdsProtocol{ +public abstract class AbstractProtocol> implements XdsProtocol { private static final Logger logger = LoggerFactory.getLogger(AbstractProtocol.class); @@ -60,7 +63,7 @@ public abstract class AbstractProtocol> implements * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request ) * K - requestId, V - StreamObserver */ - private final Map> deltaRequestObserverMap = new ConcurrentHashMap<>(); + private final Map> observeScheduledMap = new ConcurrentHashMap<>(); /** * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver ) @@ -68,17 +71,17 @@ public abstract class AbstractProtocol> implements */ private final Map> streamResult = new ConcurrentHashMap<>(); - /** - * Store consumers for Observers ( will consume message produced by Delta-ADS ) - * K - requestId, V - Consumer - */ - private final Map> consumers = new ConcurrentHashMap<>(); + private final ScheduledExecutorService pollingExecutor; - protected final AtomicLong requestId = new AtomicLong(0); + private final int pollingTimeout; - public AbstractProtocol(XdsChannel xdsChannel, Node node) { + protected final static AtomicLong requestId = new AtomicLong(0); + + public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { this.xdsChannel = xdsChannel; this.node = node; + this.pollingExecutor = new ScheduledThreadPoolExecutor(pollingPoolSize, new NamedThreadFactory("Dubbo-registry-xds")); + this.pollingTimeout = pollingTimeout; } /** @@ -91,6 +94,7 @@ public abstract class AbstractProtocol> implements @Override public T getResource(Set resourceNames) { long request = requestId.getAndIncrement(); + resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames; // Store Request Parameter, which will be used for ACK requestParam.put(request, resourceNames); @@ -114,7 +118,7 @@ public abstract class AbstractProtocol> implements return null; } finally { // close observer - requestObserver.onCompleted(); + //requestObserver.onCompleted(); // remove temp streamResult.remove(request); @@ -126,6 +130,7 @@ public abstract class AbstractProtocol> implements @Override public long observeResource(Set resourceNames, Consumer consumer) { long request = requestId.getAndIncrement(); + resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames; // Store Request Parameter, which will be used for ACK requestParam.put(request, resourceNames); @@ -133,22 +138,51 @@ public abstract class AbstractProtocol> implements // call once for full data consumer.accept(getResource(resourceNames)); - consumers.put(request, consumer); - deltaRequestObserverMap.compute(request, (k, v) -> { - // create Delta-ADS observer - v= xdsChannel.observeDeltaDiscoveryRequest(new DeltaResponseObserver(request)); + // channel reused + StreamObserver requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request)); + requestObserverMap.put(request, requestObserver); + + ScheduledFuture scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> { + try { + // origin request, may changed by updateObserve + Set names = requestParam.get(request); + + // use future to get async result + CompletableFuture future = new CompletableFuture<>(); + streamResult.put(request, future); + + // observer reused + StreamObserver observer = requestObserverMap.get(request); + + // send request to control panel + observer.onNext(buildDiscoveryRequest(names)); + + try { + // get result + consumer.accept(future.get()); + } catch (InterruptedException | ExecutionException e) { + logger.error("Error occur when request control panel."); + } finally { + // close observer + //requestObserver.onCompleted(); + + // remove temp + streamResult.remove(request); + } + } catch (Throwable t) { + logger.error("Error when requesting observe data. Type: " + getTypeUrl(), t); + } + }, pollingTimeout, pollingTimeout, TimeUnit.SECONDS); + + observeScheduledMap.put(request, scheduledFuture); - // send observe request - v.onNext(buildDeltaDiscoveryRequest(resourceNames)); - return v; - }); return request; } @Override public void updateObserve(long request, Set resourceNames) { // send difference in resourceNames - deltaRequestObserverMap.get(request).onNext(buildDeltaDiscoveryRequest(request, resourceNames)); + requestParam.put(request, resourceNames); } protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) { @@ -163,50 +197,14 @@ public abstract class AbstractProtocol> implements // for ACK return DiscoveryRequest.newBuilder() .setNode(node) - .setTypeUrl(getTypeUrl()) - .addAllResourceNames(resourceNames) + .setTypeUrl(response.getTypeUrl()) .setVersionInfo(response.getVersionInfo()) .setResponseNonce(response.getNonce()) .build(); } - protected DeltaDiscoveryRequest buildDeltaDiscoveryRequest(Set resourceNames) { - return DeltaDiscoveryRequest.newBuilder() - .setNode(node) - .setTypeUrl(getTypeUrl()) - .addAllResourceNamesSubscribe(resourceNames) - .build(); - } - - protected DeltaDiscoveryRequest buildDeltaDiscoveryRequest(long request, Set resourceNames) { - // compare with previous - Set previous = requestParam.get(request); - Set unsubscribe = new HashSet(previous) {{ - removeAll(resourceNames); - }}; - requestParam.put(request, resourceNames); - return DeltaDiscoveryRequest.newBuilder() - .setNode(node) - .setTypeUrl(getTypeUrl()) - .addAllResourceNamesUnsubscribe(unsubscribe) - .addAllResourceNamesSubscribe(resourceNames) - .build(); - } - - private DeltaDiscoveryRequest buildDeltaDiscoveryRequest(Set resourceNames, DeltaDiscoveryResponse response) { - // for ACK - return DeltaDiscoveryRequest.newBuilder() - .setNode(node) - .setTypeUrl(getTypeUrl()) - .addAllResourceNamesSubscribe(resourceNames) - .setResponseNonce(response.getNonce()) - .build(); - } - protected abstract T decodeDiscoveryResponse(DiscoveryResponse response); - protected abstract S decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, S previous); - private class ResponseObserver implements StreamObserver { private final long requestId; @@ -216,36 +214,18 @@ public abstract class AbstractProtocol> implements @Override public void onNext(DiscoveryResponse value) { + logger.info("receive notification from xds server, type: " + getTypeUrl() + " requestId: " + requestId); T result = decodeDiscoveryResponse(value); - requestObserverMap.get(requestId).onNext(buildDiscoveryRequest(requestParam.get(requestId), value)); - streamResult.get(requestId).complete(result); - } - - @Override - public void onError(Throwable t) { - logger.error("xDS Client received error message! detail:", t); - } - - @Override - public void onCompleted() { - // ignore - } - } - - private class DeltaResponseObserver implements StreamObserver { - private S delta = null; - private final long requestId; - - public DeltaResponseObserver(long requestId) { - this.requestId = requestId; - } - - @Override - public void onNext(DeltaDiscoveryResponse value) { - delta = decodeDeltaDiscoveryResponse(value, delta); - T routes = delta.getResource(); - consumers.get(requestId).accept(routes); - deltaRequestObserverMap.get(requestId).onNext(buildDeltaDiscoveryRequest(requestParam.get(requestId), value)); + StreamObserver observer = requestObserverMap.get(requestId); + if (observer == null) { + return; + } + observer.onNext(buildDiscoveryRequest(Collections.emptySet(), value)); + CompletableFuture future = streamResult.get(requestId); + if (future == null) { + return; + } + future.complete(result); } @Override diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java index 7476c00aab..8f48044456 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java @@ -31,9 +31,7 @@ import io.envoyproxy.envoy.config.core.v3.Node; import io.envoyproxy.envoy.config.core.v3.SocketAddress; import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; -import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; -import io.envoyproxy.envoy.service.discovery.v3.Resource; import java.util.Objects; import java.util.Set; @@ -43,8 +41,8 @@ public class EdsProtocol extends AbstractProtocol private static final Logger logger = LoggerFactory.getLogger(LdsProtocol.class); - public EdsProtocol(XdsChannel xdsChannel, Node node) { - super(xdsChannel, node); + public EdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { + super(xdsChannel, node, pollingPoolSize, pollingTimeout); } @Override @@ -65,25 +63,6 @@ public class EdsProtocol extends AbstractProtocol return new EndpointResult(); } - @Override - protected DeltaEndpoint decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, DeltaEndpoint previous) { - DeltaEndpoint deltaEndpoint = previous; - if (deltaEndpoint == null) { - deltaEndpoint = new DeltaEndpoint(); - } - if (getTypeUrl().equals(response.getTypeUrl())) { - deltaEndpoint.removeResource(response.getRemovedResourcesList()); - for (Resource resource : response.getResourcesList()) { - ClusterLoadAssignment unpackedResource = unpackClusterLoadAssignment(resource.getResource()); - if (unpackedResource == null) { - continue; - } - deltaEndpoint.addResource(resource.getName(), decodeResourceToEndpoint(unpackedResource)); - } - } - return previous; - } - private static Set decodeResourceToEndpoint(ClusterLoadAssignment resource) { return resource.getEndpointsList().stream() .flatMap((e) -> e.getLbEndpointsList().stream()) diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java index 3116efa850..d509134a17 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java @@ -30,10 +30,9 @@ import io.envoyproxy.envoy.config.listener.v3.Filter; import io.envoyproxy.envoy.config.listener.v3.Listener; import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager; import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.Rds; -import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; -import io.envoyproxy.envoy.service.discovery.v3.Resource; +import java.util.Collections; import java.util.Objects; import java.util.Set; import java.util.function.Consumer; @@ -43,8 +42,8 @@ public class LdsProtocol extends AbstractProtocol private static final Logger logger = LoggerFactory.getLogger(LdsProtocol.class); - public LdsProtocol(XdsChannel xdsChannel, Node node) { - super(xdsChannel, node); + public LdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { + super(xdsChannel, node, pollingPoolSize, pollingTimeout); } @Override @@ -57,7 +56,7 @@ public class LdsProtocol extends AbstractProtocol } public void observeListeners(Consumer consumer) { - observeResource(null,consumer); + observeResource(Collections.emptySet(), consumer); } @Override @@ -73,25 +72,6 @@ public class LdsProtocol extends AbstractProtocol return new ListenerResult(); } - @Override - protected DeltaListener decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, DeltaListener previous) { - DeltaListener deltaListener = previous; - if (deltaListener == null) { - deltaListener = new DeltaListener(); - } - if (getTypeUrl().equals(response.getTypeUrl())) { - deltaListener.removeResource(response.getRemovedResourcesList()); - for (Resource resource : response.getResourcesList()) { - Listener unpackedResource = unpackListener(resource.getResource()); - if (unpackedResource == null) { - continue; - } - deltaListener.addResource(resource.getName(), decodeResourceToListener(unpackedResource)); - } - } - return deltaListener; - } - private Set decodeResourceToListener(Listener resource) { return resource.getFilterChainsList().stream() .flatMap((e) -> e.getFiltersList().stream()) @@ -114,6 +94,9 @@ public class LdsProtocol extends AbstractProtocol private static HttpConnectionManager unpackHttpConnectionManager(Any any) { try { + if (!any.is(HttpConnectionManager.class)) { + return null; + } return any.unpack(HttpConnectionManager.class); } catch (InvalidProtocolBufferException e) { logger.error("Error occur when decode xDS response.", e); diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java index a85b2eaf22..43417d7e2a 100644 --- a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java @@ -29,9 +29,7 @@ import io.envoyproxy.envoy.config.core.v3.Node; import io.envoyproxy.envoy.config.route.v3.Route; import io.envoyproxy.envoy.config.route.v3.RouteAction; import io.envoyproxy.envoy.config.route.v3.RouteConfiguration; -import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; -import io.envoyproxy.envoy.service.discovery.v3.Resource; import java.util.HashMap; import java.util.Map; @@ -43,8 +41,8 @@ public class RdsProtocol extends AbstractProtocol { private static final Logger logger = LoggerFactory.getLogger(AbstractProtocol.class); - public RdsProtocol(XdsChannel xdsChannel, Node node) { - super(xdsChannel, node); + public RdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { + super(xdsChannel, node, pollingPoolSize, pollingTimeout); } @Override @@ -68,25 +66,6 @@ public class RdsProtocol extends AbstractProtocol { return new RouteResult(); } - @Override - protected DeltaRoute decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, DeltaRoute previous) { - DeltaRoute deltaRoute = previous; - if (deltaRoute == null) { - deltaRoute = new DeltaRoute(); - } - if (getTypeUrl().equals(response.getTypeUrl())) { - deltaRoute.removeResource(response.getRemovedResourcesList()); - for (Resource resource : response.getResourcesList()) { - RouteConfiguration unpackedResource = unpackRouteConfiguration(resource.getResource()); - if (unpackedResource == null) { - continue; - } - deltaRoute.addResource(resource.getName(), decodeResourceToListener(unpackedResource)); - } - } - return deltaRoute; - } - private static Map> decodeResourceToListener(RouteConfiguration resource) { Map> map = new HashMap<>(); resource.getVirtualHostsList() diff --git a/dubbo-registry/dubbo-registry-xds/src/main/proto/ca.proto b/dubbo-registry/dubbo-registry-xds/src/main/proto/ca.proto new file mode 100644 index 0000000000..ea6e7cd51a --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/proto/ca.proto @@ -0,0 +1,57 @@ +// Copyright Istio 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. + +syntax = "proto3"; + +import "google/protobuf/struct.proto"; + +// Keep this package for backward compatibility. +package istio.v1.auth; + +option go_package="istio.io/api/security/v1alpha1"; + +// Certificate request message. The authentication should be based on: +// 1. Bearer tokens carried in the side channel; +// 2. Client-side certificate via Mutual TLS handshake. +// Note: the service implementation is REQUIRED to verify the authenticated caller is authorize to +// all SANs in the CSR. The server side may overwrite any requested certificate field based on its +// policies. +message IstioCertificateRequest { + // PEM-encoded certificate request. + // The public key in the CSR is used to generate the certificate, + // and other fields in the generated certificate may be overwritten by the CA. + string csr = 1; + // Optional: requested certificate validity period, in seconds. + int64 validity_duration = 3; + + // $hide_from_docs + // Optional: Opaque metadata provided by the XDS node to Istio. + // Supported metadata: WorkloadName, WorkloadIP, ClusterID + google.protobuf.Struct metadata = 4; +} + +// Certificate response message. +message IstioCertificateResponse { + // PEM-encoded certificate chain. + // The leaf cert is the first element, and the root cert is the last element. + repeated string cert_chain = 1; +} + +// Service for managing certificates issued by the CA. +service IstioCertificateService { + // Using provided CSR, returns a signed certificate. + rpc CreateCertificate(IstioCertificateRequest) + returns (IstioCertificateResponse) { + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner new file mode 100644 index 0000000000..0a732eedb9 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner @@ -0,0 +1 @@ +istio=org.apache.dubbo.registry.xds.istio.IstioCitadelCertificateSigner \ No newline at end of file diff --git a/pom.xml b/pom.xml index a7bf7f6f68..f0cd60197f 100644 --- a/pom.xml +++ b/pom.xml @@ -282,6 +282,8 @@ **/org/apache/dubbo/common/serialize/protobuf/support/wrapper/MapValue.java, **/org/apache/dubbo/common/serialize/protobuf/support/wrapper/ThrowablePB.java, **/org/apache/dubbo/triple/TripleWrapper.java, + **/istio/v1/auth/Ca.java, + **/istio/v1/auth/IstioCertificateServiceGrpc.java,