natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Introduce cert signer for xDS, opt xDS Impl (#7934)

This commit is contained in:
Albumen Kevin 2021-06-04 16:40:40 +08:00 committed by GitHub
parent d75d3a0bd1
commit f53d8a53d7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 783 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
dubbo-dependencies-bom/pom.xml
View File

@ -161,6 +161,7 @@
<swagger_version>1.5.19</swagger_version>
<roaringbitmap_version>0.9.0</roaringbitmap_version>
<bouncycastle-bcprov_version>1.68</bouncycastle-bcprov_version>
<metrics_version>2.0.1</metrics_version>
<sofa_registry_version>5.2.0</sofa_registry_version>
<gson_version>2.8.5</gson_version>
@ -344,6 +345,22 @@
<artifactId>RoaringBitmap</artifactId>
<version>${roaringbitmap_version}</version>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk15on</artifactId>
<version>${bouncycastle-bcprov_version}</version>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcpkix-jdk15on</artifactId>
<version>${bouncycastle-bcprov_version}</version>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-ext-jdk15on</artifactId>
<version>${bouncycastle-bcprov_version}</version>
</dependency>
<!-- Common Annotations API -->
<dependency>
<groupId>javax.annotation</groupId>

46
dubbo-registry/dubbo-registry-xds/pom.xml
View File

@ -53,6 +53,11 @@
<artifactId>grpc-stub</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
</dependency>
<dependency>
<groupId>io.envoyproxy.controlplane</groupId>
<artifactId>api</artifactId>
@ -63,6 +68,47 @@
<artifactId>protobuf-java-util</artifactId>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk15on</artifactId>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcpkix-jdk15on</artifactId>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-ext-jdk15on</artifactId>
</dependency>
</dependencies>
<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.6.2</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<configuration>
<protocArtifact>com.google.protobuf:protoc:3.5.1-1:exe:${os.detected.classifier}</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.14.0:exe:${os.detected.classifier}</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

52
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java Normal file
View File

@ -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;
}
}
}

21
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java Normal file
View File

@ -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();
}

22
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java
View File

@ -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<ServiceInstance> changedToInstances(String serviceName, Collection<Endpoint> endpoints) {
List<ServiceInstance> 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;

12
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java
View File

@ -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;
}
}

249
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java Normal file
View File

@ -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<String> 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<Ca.IstioCertificateResponse> generateResponseObserver(CountDownLatch countDownLatch, StringBuffer publicKeyBuilder, AtomicBoolean failed) {
return new StreamObserver<Ca.IstioCertificateResponse>() {
@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;
}
}

53
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java Normal file
View File

@ -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";
}

121
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java Normal file
View File

@ -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;
}
}

43
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
View File

@ -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<String, Long> 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<String> 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) {

32
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java
View File

@ -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<DeltaDiscoveryRequest> observeDeltaDiscoveryRequest(StreamObserver<DeltaDiscoveryResponse> observer) {

156
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
View File

@ -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<T, S extends DeltaResource<T>> implements XdsProtocol<T>{
public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
private static final Logger logger = LoggerFactory.getLogger(AbstractProtocol.class);
@ -60,7 +63,7 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
* Store Delta-ADS Request Observer ( StreamObserver in Streaming Request )
* K - requestId, V - StreamObserver
*/
private final Map<Long, StreamObserver<DeltaDiscoveryRequest>> deltaRequestObserverMap = new ConcurrentHashMap<>();
private final Map<Long, ScheduledFuture<?>> observeScheduledMap = new ConcurrentHashMap<>();
/**
* Store CompletableFuture for Request ( used to fetch async result in ResponseObserver )
@ -68,17 +71,17 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
*/
private final Map<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
/**
* Store consumers for Observers ( will consume message produced by Delta-ADS )
* K - requestId, V - Consumer
*/
private final Map<Long, Consumer<T>> 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<T, S extends DeltaResource<T>> implements
@Override
public T getResource(Set<String> 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<T, S extends DeltaResource<T>> implements
return null;
} finally {
// close observer
requestObserver.onCompleted();
//requestObserver.onCompleted();
// remove temp
streamResult.remove(request);
@ -126,6 +130,7 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
@Override
public long observeResource(Set<String> resourceNames, Consumer<T> 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<T, S extends DeltaResource<T>> 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<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
requestObserverMap.put(request, requestObserver);
ScheduledFuture<?> scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> {
try {
// origin request, may changed by updateObserve
Set<String> names = requestParam.get(request);
// use future to get async result
CompletableFuture<T> future = new CompletableFuture<>();
streamResult.put(request, future);
// observer reused
StreamObserver<DiscoveryRequest> 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<String> resourceNames) {
// send difference in resourceNames
deltaRequestObserverMap.get(request).onNext(buildDeltaDiscoveryRequest(request, resourceNames));
requestParam.put(request, resourceNames);
}
protected DiscoveryRequest buildDiscoveryRequest(Set<String> resourceNames) {
@ -163,50 +197,14 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> 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<String> resourceNames) {
return DeltaDiscoveryRequest.newBuilder()
.setNode(node)
.setTypeUrl(getTypeUrl())
.addAllResourceNamesSubscribe(resourceNames)
.build();
}
protected DeltaDiscoveryRequest buildDeltaDiscoveryRequest(long request, Set<String> resourceNames) {
// compare with previous
Set<String> previous = requestParam.get(request);
Set<String> unsubscribe = new HashSet<String>(previous) {{
removeAll(resourceNames);
}};
requestParam.put(request, resourceNames);
return DeltaDiscoveryRequest.newBuilder()
.setNode(node)
.setTypeUrl(getTypeUrl())
.addAllResourceNamesUnsubscribe(unsubscribe)
.addAllResourceNamesSubscribe(resourceNames)
.build();
}
private DeltaDiscoveryRequest buildDeltaDiscoveryRequest(Set<String> 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<DiscoveryResponse> {
private final long requestId;
@ -216,36 +214,18 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> 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<DeltaDiscoveryResponse> {
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<DiscoveryRequest> observer = requestObserverMap.get(requestId);
if (observer == null) {
return;
}
observer.onNext(buildDiscoveryRequest(Collections.emptySet(), value));
CompletableFuture<T> future = streamResult.get(requestId);
if (future == null) {
return;
}
future.complete(result);
}
@Override

25
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java
View File

@ -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<EndpointResult, DeltaEndpoint>
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<EndpointResult, DeltaEndpoint>
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<Endpoint> decodeResourceToEndpoint(ClusterLoadAssignment resource) {
return resource.getEndpointsList().stream()
.flatMap((e) -> e.getLbEndpointsList().stream())

31
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
View File

@ -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<ListenerResult, DeltaListener>
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<ListenerResult, DeltaListener>
}
public void observeListeners(Consumer<ListenerResult> consumer) {
observeResource(null,consumer);
observeResource(Collections.emptySet(), consumer);
}
@Override
@ -73,25 +72,6 @@ public class LdsProtocol extends AbstractProtocol<ListenerResult, DeltaListener>
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<String> decodeResourceToListener(Listener resource) {
return resource.getFilterChainsList().stream()
.flatMap((e) -> e.getFiltersList().stream())
@ -114,6 +94,9 @@ public class LdsProtocol extends AbstractProtocol<ListenerResult, DeltaListener>
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);

25
dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java
View File

@ -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<RouteResult, DeltaRoute> {
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<RouteResult, DeltaRoute> {
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<String, Set<String>> decodeResourceToListener(RouteConfiguration resource) {
Map<String, Set<String>> map = new HashMap<>();
resource.getVirtualHostsList()

57
dubbo-registry/dubbo-registry-xds/src/main/proto/ca.proto Normal file
View File

@ -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) {
}
}

1
dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner Normal file
View File

@ -0,0 +1 @@
istio=org.apache.dubbo.registry.xds.istio.IstioCitadelCertificateSigner

2
pom.xml
View File

@ -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,
</excludes>
</configuration>
<goals>