[Part 2] | TimeLock shoots itself (#4983)

* TimeLock corruption pingers

* Refactor

* Redesign

* Refactor

* Wire up corruption pingers through TL

* Minor refactor

* Refactor

* Rewire

* Refactor

* Minor refactor

timelock-agent/src/main/java/com/palantir/atlasdb/timelock/paxos/LeadershipComponents.java

@@ -30,11 +30,15 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
 import com.google.common.io.Closer;
+import com.palantir.atlasdb.timelock.corruption.CorruptionHealthCheck;
+import com.palantir.atlasdb.timelock.corruption.LocalCorruptionDetector;
+import com.palantir.atlasdb.timelock.corruption.RemoteCorruptionDetector;
 import com.palantir.atlasdb.timelock.paxos.NetworkClientFactories.Factory;
 import com.palantir.leader.LeaderElectionService;
 import com.palantir.leader.NotCurrentLeaderException;
 import com.palantir.leader.proxy.AwaitingLeadershipProxy;
 import com.palantir.paxos.Client;
+import com.palantir.timelock.corruption.TimeLockCorruptionNotifier;
 import com.palantir.timelock.paxos.HealthCheckPinger;
 import com.palantir.timelock.paxos.LeaderPingHealthCheck;
 import com.palantir.timelock.paxos.NamespaceTracker;
@@ -48,12 +52,15 @@ public class LeadershipComponents {
 
     private final Factory<LeadershipContext> leadershipContextFactory;
     private final LocalAndRemotes<HealthCheckPinger> healthCheckPingers;
+    private final List<TimeLockCorruptionNotifier> corruptionNotifiers;
 
     LeadershipComponents(
             Factory<LeadershipContext> leadershipContextFactory,
-            LocalAndRemotes<HealthCheckPinger> healthCheckPingers) {
+            LocalAndRemotes<HealthCheckPinger> healthCheckPingers,
+            List<TimeLockCorruptionNotifier> corruptionNotifiers) {
         this.leadershipContextFactory = leadershipContextFactory;
         this.healthCheckPingers = healthCheckPingers;
+        this.corruptionNotifiers = corruptionNotifiers;
     }
 
     public <T> T wrapInLeadershipProxy(Client client, Class<T> clazz, Supplier<T> delegateSupplier) {
@@ -83,6 +90,17 @@ public LeaderPingHealthCheck healthCheck(NamespaceTracker namespaceTracker) {
         return new LeaderPingHealthCheck(namespaceTracker, healthCheckPingers.all());
     }
 
+    public TimeLockCorruptionComponents timeLockCorruptionComponents() {
+        RemoteCorruptionDetector remoteCorruptionDetector = new RemoteCorruptionDetector();
+        CorruptionHealthCheck healthCheck = new CorruptionHealthCheck(ImmutableList.of(
+                LocalCorruptionDetector.create(corruptionNotifiers),
+                remoteCorruptionDetector));
+        return ImmutableTimeLockCorruptionComponents.builder()
+                .timeLockCorruptionHealthCheck(healthCheck)
+                .remoteCorruptionDetector(remoteCorruptionDetector)
+                .build();
+    }
+
     public boolean requestHostileTakeover(Client client) {
         return getOrCreateNewLeadershipContext(client).leaderElectionService().hostileTakeover();
     }

timelock-agent/src/main/java/com/palantir/atlasdb/timelock/paxos/LeadershipContextFactory.java

@@ -18,6 +18,7 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.UUID;
 
 import org.immutables.value.Value;
@@ -31,6 +32,7 @@
 import com.palantir.paxos.Client;
 import com.palantir.paxos.LeaderPinger;
 import com.palantir.paxos.PaxosLearner;
+import com.palantir.timelock.corruption.TimeLockCorruptionNotifier;
 import com.palantir.timelock.paxos.HealthCheckPinger;
 
 @Value.Immutable
@@ -106,6 +108,11 @@ public LeaderElectionHealthCheck leaderElectionHealthCheck() {
         return new LeaderElectionHealthCheck(Instant::now);
     }
 
+    @Value.Derived
+    List<TimeLockCorruptionNotifier> remoteCorruptionNotifiers() {
+        return remoteClients().getRemoteCorruptionNotifiers();
+    }
+
     @Override
     public LeadershipContext create(Client client) {
         ClientAwareComponents clientAwareComponents = ImmutableClientAwareComponents.builder()

timelock-agent/src/main/java/com/palantir/atlasdb/timelock/paxos/PaxosRemoteClients.java

@@ -37,6 +37,7 @@
 import com.palantir.paxos.LeaderPingerContext;
 import com.palantir.paxos.PaxosAcceptor;
 import com.palantir.paxos.PaxosLearner;
+import com.palantir.timelock.corruption.TimeLockCorruptionNotifier;
 import com.palantir.timelock.paxos.TimelockPaxosAcceptorRpcClient;
 import com.palantir.timelock.paxos.TimelockPaxosLearnerRpcClient;
 
@@ -145,6 +146,13 @@ public List<WithDedicatedExecutor<LeaderPingerContext<BatchPingableLeader>>> bat
         return leaderPingerContext(BatchPingableLeader.class);
     }
 
+    @Value.Derived
+    public List<TimeLockCorruptionNotifier> getRemoteCorruptionNotifiers() {
+        return createInstrumentedRemoteProxies(TimeLockCorruptionNotifier.class, true)
+                .values()
+                .collect(Collectors.toList());
+    }
+
     private <T> List<WithDedicatedExecutor<LeaderPingerContext<T>>> leaderPingerContext(Class<T> clazz) {
         return createInstrumentedRemoteProxies(clazz, false)
                 .<WithDedicatedExecutor<LeaderPingerContext<T>>>map((uri, remote) ->

timelock-agent/src/main/java/com/palantir/atlasdb/timelock/paxos/PaxosResources.java

@@ -66,7 +66,8 @@ public List<Object> resourcesForRegistration() {
     public LeadershipComponents leadershipComponents() {
         return new LeadershipComponents(
                 leadershipContextFactory(),
-                leadershipContextFactory().healthCheckPingers());
+                leadershipContextFactory().healthCheckPingers(),
+                leadershipContextFactory().remoteCorruptionNotifiers());
     }
 
     private static BatchPaxosResources batchResourcesFromComponents(LocalPaxosComponents components) {

timelock-agent/src/main/java/com/palantir/atlasdb/timelock/paxos/TimeLockCorruptionComponents.java

@@ -0,0 +1,28 @@
+/*
+ * (c) Copyright 2020 Palantir Technologies Inc. All rights reserved.
+ *
+ * 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 com.palantir.atlasdb.timelock.paxos;
+
+import org.immutables.value.Value;
+
+import com.palantir.atlasdb.timelock.corruption.CorruptionHealthCheck;
+import com.palantir.atlasdb.timelock.corruption.RemoteCorruptionDetector;
+
+@Value.Immutable
+public interface TimeLockCorruptionComponents {
+    CorruptionHealthCheck timeLockCorruptionHealthCheck();
+    RemoteCorruptionDetector remoteCorruptionDetector();
+}

timelock-agent/src/main/java/com/palantir/timelock/paxos/TimeLockAgent.java

@@ -44,13 +44,15 @@
 import com.palantir.atlasdb.timelock.adjudicate.FeedbackHandler;
 import com.palantir.atlasdb.timelock.adjudicate.HealthStatusReport;
 import com.palantir.atlasdb.timelock.adjudicate.TimeLockClientFeedbackResource;
+import com.palantir.atlasdb.timelock.corruption.CorruptionNotifierResource;
 import com.palantir.atlasdb.timelock.lock.LockLog;
 import com.palantir.atlasdb.timelock.lock.v1.ConjureLockV1Resource;
 import com.palantir.atlasdb.timelock.management.PersistentNamespaceContext;
 import com.palantir.atlasdb.timelock.management.TimeLockManagementResource;
 import com.palantir.atlasdb.timelock.paxos.ImmutableTimelockPaxosInstallationContext;
 import com.palantir.atlasdb.timelock.paxos.PaxosResources;
 import com.palantir.atlasdb.timelock.paxos.PaxosResourcesFactory;
+import com.palantir.atlasdb.timelock.paxos.TimeLockCorruptionComponents;
 import com.palantir.atlasdb.util.MetricsManager;
 import com.palantir.common.concurrent.PTExecutors;
 import com.palantir.conjure.java.api.config.service.ServicesConfigBlock;
@@ -200,6 +202,7 @@ private void createAndRegisterResources() {
         registerPaxosResource();
         registerExceptionMappers();
         registerClientFeedbackService();
+        registerTimeLockCorruptionNotifiers();
 
         namespaces = new TimelockNamespaces(
                 metricsManager,
@@ -241,6 +244,17 @@ private void registerClientFeedbackService() {
         }
     }
 
+    private void registerTimeLockCorruptionNotifiers() {
+        TimeLockCorruptionComponents corruptionComponents
+                = paxosResources.leadershipComponents().timeLockCorruptionComponents();
+        if (undertowRegistrar.isPresent()) {
+            undertowRegistrar.get().accept(
+                    CorruptionNotifierResource.undertow(corruptionComponents.remoteCorruptionDetector()));
+        } else {
+            registrar.accept(CorruptionNotifierResource.jersey(corruptionComponents.remoteCorruptionDetector()));
+        }
+    }
+
     private boolean isLeaderForClient(Client client) {
         Map<Client, HealthCheckResponse> healthCheckResponseMap = paxosResources
                 .leadershipComponents()