mirror of https://github.com/hyperledger/besu
Detect zombies backend query and abort useless processing. (#1042)
* Detect zombies backend query and abort useless processing. The timeout handler populate the request context with a boolean value indicating whether or not the query is alive, i.e the HTTP request has not expired. Backend queries are now conditioned by this value and can be stopped if needed. This PR experiments this mechanism on a reduced scope. Hence, only `eth_getLogs` backend queries are affected. - Created `BackendQuery` utility class to run a process only if the query is alive, i.e timeout not expired. - Put `AtomicBoolean` value in the `JsonRpcRequestContext` - `TimeoutHandler` sets the alive value to `false` if the timeout handler is triggered. - Updated `BlockchainQueries` to run steps depending on the value of the `AtomicBoolean` retrieved from the request context. - Added unit tests. Signed-off-by: Abdelhamid Bakhta <abdelhamid.bakhta@consensys.net>pull/1100/head
Abdelhamid Bakhta
4 years ago
committed by
GitHub
parent
dc4e2a8368
commit
903afdedd6
@ -0,0 +1,103 @@ |
||||
/* |
||||
* Copyright ConsenSys AG. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.graphql; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.handlers.IsAliveHandler; |
||||
import org.hyperledger.besu.ethereum.api.query.BlockchainQueries; |
||||
import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator; |
||||
import org.hyperledger.besu.ethereum.core.Synchronizer; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool; |
||||
import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule; |
||||
|
||||
public class GraphQLDataFetcherContextImpl implements GraphQLDataFetcherContext { |
||||
|
||||
private final BlockchainQueries blockchainQueries; |
||||
private final MiningCoordinator miningCoordinator; |
||||
private final Synchronizer synchronizer; |
||||
private final ProtocolSchedule protocolSchedule; |
||||
private final TransactionPool transactionPool; |
||||
private final IsAliveHandler isAliveHandler; |
||||
|
||||
public GraphQLDataFetcherContextImpl( |
||||
final GraphQLDataFetcherContext context, final IsAliveHandler isAliveHandler) { |
||||
this( |
||||
context.getBlockchainQueries(), |
||||
context.getProtocolSchedule(), |
||||
context.getTransactionPool(), |
||||
context.getMiningCoordinator(), |
||||
context.getSynchronizer(), |
||||
isAliveHandler); |
||||
} |
||||
|
||||
public GraphQLDataFetcherContextImpl( |
||||
final BlockchainQueries blockchainQueries, |
||||
final ProtocolSchedule protocolSchedule, |
||||
final TransactionPool transactionPool, |
||||
final MiningCoordinator miningCoordinator, |
||||
final Synchronizer synchronizer) { |
||||
this( |
||||
blockchainQueries, |
||||
protocolSchedule, |
||||
transactionPool, |
||||
miningCoordinator, |
||||
synchronizer, |
||||
new IsAliveHandler(true)); |
||||
} |
||||
|
||||
public GraphQLDataFetcherContextImpl( |
||||
final BlockchainQueries blockchainQueries, |
||||
final ProtocolSchedule protocolSchedule, |
||||
final TransactionPool transactionPool, |
||||
final MiningCoordinator miningCoordinator, |
||||
final Synchronizer synchronizer, |
||||
final IsAliveHandler isAliveHandler) { |
||||
this.blockchainQueries = blockchainQueries; |
||||
this.protocolSchedule = protocolSchedule; |
||||
this.miningCoordinator = miningCoordinator; |
||||
this.synchronizer = synchronizer; |
||||
this.transactionPool = transactionPool; |
||||
this.isAliveHandler = isAliveHandler; |
||||
} |
||||
|
||||
@Override |
||||
public TransactionPool getTransactionPool() { |
||||
return transactionPool; |
||||
} |
||||
|
||||
@Override |
||||
public BlockchainQueries getBlockchainQueries() { |
||||
return blockchainQueries; |
||||
} |
||||
|
||||
@Override |
||||
public MiningCoordinator getMiningCoordinator() { |
||||
return miningCoordinator; |
||||
} |
||||
|
||||
@Override |
||||
public Synchronizer getSynchronizer() { |
||||
return synchronizer; |
||||
} |
||||
|
||||
@Override |
||||
public ProtocolSchedule getProtocolSchedule() { |
||||
return protocolSchedule; |
||||
} |
||||
|
||||
@Override |
||||
public IsAliveHandler getIsAliveHandler() { |
||||
return isAliveHandler; |
||||
} |
||||
} |
||||
@ -0,0 +1,53 @@ |
||||
/* |
||||
* Copyright ConsenSys AG. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.handlers; |
||||
|
||||
import org.hyperledger.besu.ethereum.eth.manager.EthScheduler; |
||||
|
||||
import java.time.Duration; |
||||
import java.util.concurrent.atomic.AtomicBoolean; |
||||
import java.util.function.Supplier; |
||||
|
||||
public class IsAliveHandler implements Supplier<Boolean> { |
||||
|
||||
private final AtomicBoolean alive; |
||||
|
||||
public IsAliveHandler(final boolean alive) { |
||||
this(new AtomicBoolean(alive)); |
||||
} |
||||
|
||||
public IsAliveHandler(final AtomicBoolean alive) { |
||||
this.alive = alive; |
||||
} |
||||
|
||||
public IsAliveHandler(final EthScheduler ethScheduler, final long timeoutSec) { |
||||
this(ethScheduler, new AtomicBoolean(true), timeoutSec); |
||||
} |
||||
|
||||
public IsAliveHandler( |
||||
final EthScheduler ethScheduler, final AtomicBoolean alive, final long timeoutSec) { |
||||
this.alive = alive; |
||||
ethScheduler.scheduleFutureTask(this::triggerTimeout, Duration.ofSeconds(timeoutSec)); |
||||
} |
||||
|
||||
private void triggerTimeout() { |
||||
alive.set(false); |
||||
} |
||||
|
||||
@Override |
||||
public Boolean get() { |
||||
return alive.get(); |
||||
} |
||||
} |
||||
@ -0,0 +1,21 @@ |
||||
/* |
||||
* Copyright ConsenSys AG. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.handlers; |
||||
|
||||
public class RpcMethodTimeoutException extends RuntimeException { |
||||
public RpcMethodTimeoutException() { |
||||
super("Timeout expired"); |
||||
} |
||||
} |
||||
@ -0,0 +1,54 @@ |
||||
/* |
||||
* Copyright ConsenSys AG. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.query; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.handlers.RpcMethodTimeoutException; |
||||
|
||||
import java.util.Optional; |
||||
import java.util.concurrent.Callable; |
||||
import java.util.function.Supplier; |
||||
|
||||
import org.apache.logging.log4j.LogManager; |
||||
import org.apache.logging.log4j.Logger; |
||||
|
||||
public class BackendQuery { |
||||
private static final Logger LOG = LogManager.getLogger(); |
||||
|
||||
public static <T> T runIfAlive(final Callable<T> task, final Supplier<Boolean> alive) |
||||
throws Exception { |
||||
return runIfAlive(Optional.empty(), task, alive); |
||||
} |
||||
|
||||
public static <T> T runIfAlive( |
||||
final String taskName, final Callable<T> task, final Supplier<Boolean> alive) |
||||
throws Exception { |
||||
return runIfAlive(Optional.ofNullable(taskName), task, alive); |
||||
} |
||||
|
||||
public static <T> T runIfAlive( |
||||
final Optional<String> taskName, final Callable<T> task, final Supplier<Boolean> alive) |
||||
throws Exception { |
||||
if (!alive.get()) { |
||||
LOG.warn( |
||||
"Zombie backend query detected [ {} ], aborting process.", taskName.orElse("unnamed")); |
||||
throw new RpcMethodTimeoutException(); |
||||
} |
||||
return task.call(); |
||||
} |
||||
|
||||
public static void stopIfExpired(final Supplier<Boolean> alive) throws Exception { |
||||
runIfAlive(() -> null, alive); |
||||
} |
||||
} |
||||
@ -0,0 +1,74 @@ |
||||
/* |
||||
* Copyright ConsenSys AG. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.query; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
||||
|
||||
import java.util.Arrays; |
||||
import java.util.Collection; |
||||
import java.util.function.Supplier; |
||||
|
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.junit.runners.Parameterized; |
||||
import org.junit.runners.Parameterized.Parameters; |
||||
|
||||
@RunWith(Parameterized.class) |
||||
public class BackendQueryTest<T> { |
||||
|
||||
private final Supplier<Boolean> alive; |
||||
private final Object wantReturn; |
||||
private final boolean wantException; |
||||
private final Class<T> wantExceptionClass; |
||||
private final String wantExceptionMessage; |
||||
|
||||
public BackendQueryTest( |
||||
final Supplier<Boolean> alive, |
||||
final Object wantReturn, |
||||
final boolean wantException, |
||||
final Class<T> wantExceptionClass, |
||||
final String wantExceptionMessage) { |
||||
this.alive = alive; |
||||
this.wantReturn = wantReturn; |
||||
this.wantException = wantException; |
||||
this.wantExceptionClass = wantExceptionClass; |
||||
this.wantExceptionMessage = wantExceptionMessage; |
||||
} |
||||
|
||||
@Parameters |
||||
public static Collection<Object[]> data() { |
||||
return Arrays.asList( |
||||
new Object[][] { |
||||
{supplierOf(false), null, true, RuntimeException.class, "Timeout expired"}, |
||||
{supplierOf(true), "expected return", false, null, null} |
||||
}); |
||||
} |
||||
|
||||
private static Supplier<Boolean> supplierOf(final boolean val) { |
||||
return () -> val; |
||||
} |
||||
|
||||
@Test |
||||
public void test() throws Exception { |
||||
if (wantException) { |
||||
assertThatThrownBy(() -> BackendQuery.runIfAlive(() -> wantReturn, alive)) |
||||
.isInstanceOf(wantExceptionClass) |
||||
.hasMessage(wantExceptionMessage); |
||||
} else { |
||||
assertThat(BackendQuery.runIfAlive(() -> wantReturn, alive)).isEqualTo(wantReturn); |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue