#19657

Committed 23 Jan 2025 12:00AM CUT coverage: 88.581% (+0.005%) from 88.576%

Build # #19657

Build Type

push

github

Committed by

ejona86

Commit Message

xds: Fix fallback test FakeClock TSAN failure

d65d3942e increased the test speed of
connect_then_mainServerDown_fallbackServerUp by using FakeClock.
However, it introduced a data race because FakeClock is not thread-safe.
This change injects a single thread for gRPC callbacks such that
syncContext is run on a thread under the test's control.

A simpler approach would be to expose syncContext from XdsClientImpl for
testing. However, this test is in a different package and I wanted to
avoid adding a public method.

```
  Read of size 8 at 0x00008dec9d50 by thread T25:
    #0 io.grpc.internal.FakeClock$ScheduledExecutorImpl.schedule(Lio/grpc/internal/FakeClock$ScheduledTask;JLjava/util/concurrent/TimeUnit;)V FakeClock.java:140
    #1 io.grpc.internal.FakeClock$ScheduledExecutorImpl.schedule(Ljava/lang/Runnable;JLjava/util/concurrent/TimeUnit;)Ljava/util/concurrent/ScheduledFuture; FakeClock.java:150
    #2 io.grpc.SynchronizationContext.schedule(Ljava/lang/Runnable;JLjava/util/concurrent/TimeUnit;Ljava/util/concurrent/ScheduledExecutorService;)Lio/grpc/SynchronizationContext$ScheduledHandle; SynchronizationContext.java:153
    #3 io.grpc.xds.client.ControlPlaneClient$AdsStream.handleRpcStreamClosed(Lio/grpc/Status;)V ControlPlaneClient.java:491
    #4 io.grpc.xds.client.ControlPlaneClient$AdsStream.lambda$onStatusReceived$0(Lio/grpc/Status;)V ControlPlaneClient.java:429
    #5 io.grpc.xds.client.ControlPlaneClient$AdsStream$$Lambda+0x00000001004a95d0.run()V ??
    #6 io.grpc.SynchronizationContext.drain()V SynchronizationContext.java:96
    #7 io.grpc.SynchronizationContext.execute(Ljava/lang/Runnable;)V SynchronizationContext.java:128
    #8 io.grpc.xds.client.ControlPlaneClient$AdsStream.onStatusReceived(Lio/grpc/Status;)V ControlPlaneClient.java:428
    #9 io.grpc.xds.GrpcXdsTransportFactory$EventHandlerToCallListenerAdapter.onClose(Lio/grpc/Status;Lio/grpc/Metadata;)V GrpcXdsTransportFactory.java:149
    #10 io.grpc.PartialForwardingClientCallListen... (continued)

Run Details

33721 of 38068 relevant lines covered (88.58%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.55

/../xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java

/*
 * Copyright 2020 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds.internal.security.certprovider;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import io.grpc.Status;
import io.grpc.xds.internal.security.Closeable;
import java.security.PrivateKey;
import java.security.cert.X509Certificate;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * A plug-in that provides certificates required by the xDS security component and created
 * using the certificate-provider config from the xDS server.
 *
 * <p>We may move this out of the internal package and make this an official API in the future.
 *
 * <p>The plugin fetches certificates - root and optionally identity cert - required by xDS
 * security.
 */
public abstract class CertificateProvider implements Closeable {

  /** A watcher is registered to receive certificate updates. */
  public interface Watcher {
    void updateCertificate(PrivateKey key, List<X509Certificate> certChain);

    void updateTrustedRoots(List<X509Certificate> trustedRoots);

    void updateSpiffeTrustMap(Map<String, List<X509Certificate>> spiffeTrustMap);

    void onError(Status errorStatus);
  }

  @VisibleForTesting
  public static final class DistributorWatcher implements Watcher {
    private PrivateKey privateKey;
    private List<X509Certificate> certChain;
    private List<X509Certificate> trustedRoots;
    private Map<String, List<X509Certificate>> spiffeTrustMap;

    @VisibleForTesting
    final Set<Watcher> downstreamWatchers = new HashSet<>();

    synchronized void addWatcher(Watcher watcher) {
      downstreamWatchers.add(watcher);
      if (privateKey != null && certChain != null) {
        sendLastCertificateUpdate(watcher);
      }
      if (trustedRoots != null) {
        sendLastTrustedRootsUpdate(watcher);
      }
      if (spiffeTrustMap != null) {
        sendLastSpiffeTrustMapUpdate(watcher);
      }
    }

    synchronized void removeWatcher(Watcher watcher) {
      downstreamWatchers.remove(watcher);
    }

    @VisibleForTesting public Set<Watcher> getDownstreamWatchers() {
      return Collections.unmodifiableSet(downstreamWatchers);
    }

    private void sendLastCertificateUpdate(Watcher watcher) {
      watcher.updateCertificate(privateKey, certChain);
    }

    private void sendLastTrustedRootsUpdate(Watcher watcher) {
      watcher.updateTrustedRoots(trustedRoots);
    }

    private void sendLastSpiffeTrustMapUpdate(Watcher watcher) {
      watcher.updateSpiffeTrustMap(spiffeTrustMap);
    }

    @Override
    public synchronized void updateCertificate(PrivateKey key, List<X509Certificate> certChain) {
      checkNotNull(key, "key");
      checkNotNull(certChain, "certChain");
      privateKey = key;
      this.certChain = certChain;
      for (Watcher watcher : downstreamWatchers) {
        sendLastCertificateUpdate(watcher);
      }
    }

    @Override
    public synchronized void updateTrustedRoots(List<X509Certificate> trustedRoots) {
      checkNotNull(trustedRoots, "trustedRoots");
      this.trustedRoots = trustedRoots;
      for (Watcher watcher : downstreamWatchers) {
        sendLastTrustedRootsUpdate(watcher);
      }
    }

    @Override
    public void updateSpiffeTrustMap(Map<String, List<X509Certificate>> spiffeTrustMap) {
      this.spiffeTrustMap = spiffeTrustMap;
      for (Watcher watcher : downstreamWatchers) {
        sendLastSpiffeTrustMapUpdate(watcher);
      }
    }

    @Override
    public synchronized void onError(Status errorStatus) {
      for (Watcher watcher : downstreamWatchers) {
        watcher.onError(errorStatus);
      }
    }

    X509Certificate getLastIdentityCert() {
      if (certChain != null && !certChain.isEmpty()) {
        return certChain.get(0);
      }
      return null;
    }

    void close() {
      downstreamWatchers.clear();
      clearValues();
    }

    void clearValues() {
      privateKey = null;
      certChain = null;
      trustedRoots = null;
    }
  }

  /**
   * Concrete subclasses will call this to register the {@link Watcher}.
   *
   * @param watcher to register
   * @param notifyCertUpdates if true, the provider is required to call the watcher’s
   *     updateCertificate method. Implies the Provider is capable of minting certificates.
   *     Used by server-side and mTLS client-side. Note the Provider is always required
   *     to call updateTrustedRoots to provide trusted-root updates.
   */
  protected CertificateProvider(DistributorWatcher watcher, boolean notifyCertUpdates) {
    this.watcher = watcher;
    this.notifyCertUpdates = notifyCertUpdates;
  }

  /** Releases all resources and stop cert refreshes and watcher updates. */
  @Override
  public abstract void close();

  /** Starts the async cert refresh and watcher update cycle. */
  public abstract void start();

  private final DistributorWatcher watcher;
  private final boolean notifyCertUpdates;

  public DistributorWatcher getWatcher() {
    return watcher;
  }

  public boolean isNotifyCertUpdates() {
    return notifyCertUpdates;
  }


}

1	/*
2	* Copyright 2020 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.xds.internal.security.certprovider;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import io.grpc.Status;
23	import io.grpc.xds.internal.security.Closeable;
24	import java.security.PrivateKey;
25	import java.security.cert.X509Certificate;
26	import java.util.Collections;
27	import java.util.HashSet;
28	import java.util.List;
29	import java.util.Map;
30	import java.util.Set;
31
32	/**
33	* A plug-in that provides certificates required by the xDS security component and created
34	* using the certificate-provider config from the xDS server.
35	*
36	* <p>We may move this out of the internal package and make this an official API in the future.
37	*
38	* <p>The plugin fetches certificates - root and optionally identity cert - required by xDS
39	* security.
40	*/
41	public abstract class CertificateProvider implements Closeable {
42
43	/** A watcher is registered to receive certificate updates. */
44	public interface Watcher {
45	void updateCertificate(PrivateKey key, List<X509Certificate> certChain);
46
47	void updateTrustedRoots(List<X509Certificate> trustedRoots);
48
49	void updateSpiffeTrustMap(Map<String, List<X509Certificate>> spiffeTrustMap);
50
51	void onError(Status errorStatus);
52	}
53
54	@VisibleForTesting
55	public static final class DistributorWatcher implements Watcher {	1✔
56	private PrivateKey privateKey;
57	private List<X509Certificate> certChain;
58	private List<X509Certificate> trustedRoots;
59	private Map<String, List<X509Certificate>> spiffeTrustMap;
60
61	@VisibleForTesting	1✔
62	final Set<Watcher> downstreamWatchers = new HashSet<>();
63
64	synchronized void addWatcher(Watcher watcher) {
65	downstreamWatchers.add(watcher);	1✔
66	if (privateKey != null && certChain != null) {	1✔
67	sendLastCertificateUpdate(watcher);	1✔
68	}
69	if (trustedRoots != null) {	1✔
70	sendLastTrustedRootsUpdate(watcher);	1✔
71	}
72	if (spiffeTrustMap != null) {	1✔
73	sendLastSpiffeTrustMapUpdate(watcher);	×
74	}
75	}	1✔
76
77	synchronized void removeWatcher(Watcher watcher) {
78	downstreamWatchers.remove(watcher);	1✔
79	}	1✔
80
81	@VisibleForTesting public Set<Watcher> getDownstreamWatchers() {
82	return Collections.unmodifiableSet(downstreamWatchers);	1✔
83	}
84
85	private void sendLastCertificateUpdate(Watcher watcher) {
86	watcher.updateCertificate(privateKey, certChain);	1✔
87	}	1✔
88
89	private void sendLastTrustedRootsUpdate(Watcher watcher) {
90	watcher.updateTrustedRoots(trustedRoots);	1✔
91	}	1✔
92
93	private void sendLastSpiffeTrustMapUpdate(Watcher watcher) {
94	watcher.updateSpiffeTrustMap(spiffeTrustMap);	1✔
95	}	1✔
96
97	@Override
98	public synchronized void updateCertificate(PrivateKey key, List<X509Certificate> certChain) {
99	checkNotNull(key, "key");	1✔
100	checkNotNull(certChain, "certChain");	1✔
101	privateKey = key;	1✔
102	this.certChain = certChain;	1✔
103	for (Watcher watcher : downstreamWatchers) {	1✔
104	sendLastCertificateUpdate(watcher);	1✔
105	}	1✔
106	}	1✔
107
108	@Override
109	public synchronized void updateTrustedRoots(List<X509Certificate> trustedRoots) {
110	checkNotNull(trustedRoots, "trustedRoots");	1✔
111	this.trustedRoots = trustedRoots;	1✔
112	for (Watcher watcher : downstreamWatchers) {	1✔
113	sendLastTrustedRootsUpdate(watcher);	1✔
114	}	1✔
115	}	1✔
116
117	@Override
118	public void updateSpiffeTrustMap(Map<String, List<X509Certificate>> spiffeTrustMap) {
119	this.spiffeTrustMap = spiffeTrustMap;	1✔
120	for (Watcher watcher : downstreamWatchers) {	1✔
121	sendLastSpiffeTrustMapUpdate(watcher);	1✔
122	}	1✔
123	}	1✔
124
125	@Override
126	public synchronized void onError(Status errorStatus) {
127	for (Watcher watcher : downstreamWatchers) {	1✔
128	watcher.onError(errorStatus);	1✔
129	}	1✔
130	}	1✔
131
132	X509Certificate getLastIdentityCert() {
133	if (certChain != null && !certChain.isEmpty()) {	1✔
134	return certChain.get(0);	1✔
135	}
136	return null;	1✔
137	}
138
139	void close() {
140	downstreamWatchers.clear();	1✔
141	clearValues();	1✔
142	}	1✔
143
144	void clearValues() {
145	privateKey = null;	1✔
146	certChain = null;	1✔
147	trustedRoots = null;	1✔
148	}	1✔
149	}
150
151	/**
152	* Concrete subclasses will call this to register the {@link Watcher}.
153	*
154	* @param watcher to register
155	* @param notifyCertUpdates if true, the provider is required to call the watcher’s
156	* updateCertificate method. Implies the Provider is capable of minting certificates.
157	* Used by server-side and mTLS client-side. Note the Provider is always required
158	* to call updateTrustedRoots to provide trusted-root updates.
159	*/
160	protected CertificateProvider(DistributorWatcher watcher, boolean notifyCertUpdates) {	1✔
161	this.watcher = watcher;	1✔
162	this.notifyCertUpdates = notifyCertUpdates;	1✔
163	}	1✔
164
165	/** Releases all resources and stop cert refreshes and watcher updates. */
166	@Override
167	public abstract void close();
168
169	/** Starts the async cert refresh and watcher update cycle. */
170	public abstract void start();
171
172	private final DistributorWatcher watcher;
173	private final boolean notifyCertUpdates;
174
175	public DistributorWatcher getWatcher() {
176	return watcher;	1✔
177	}
178
179	public boolean isNotifyCertUpdates() {
180	return notifyCertUpdates;	×
181	}
182
183
184	}

grpc / grpc-java / #19657

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous