Experimental flag to disable parallel processing of keystore config files

CHANGELOG.md

@@ -5,7 +5,9 @@
 
 ### Bugs Fixed
 - Upgrade jackson and vertx to upgrade snakeyaml to 2.0 to fix CVE-2022-1471
-- Fixed handling of very large number of signing metadata files with Hashicorp connection (30000+). [#736](https://github.com/ConsenSys/web3signer/issues/736)
+- Fixed handling of very large number (30,000+) of signing metadata files with Hashicorp connection by introducing 
+experimental flag to disable parallel processing `--Xmetadata-files-parallel-processing-enabled`. 
+[#794](https://github.com/ConsenSys/web3signer/pull/794)
 - Fixed startup error with web3signer where openAPI spec cannot be loaded [#772](https://github.com/ConsenSys/web3signer/issues/772)
 
 ---

commandline/src/main/java/tech/pegasys/web3signer/commandline/Web3SignerBaseCommand.java

@@ -191,6 +191,14 @@ public class Web3SignerBaseCommand implements BaseConfig, Runnable {
       description = "Enable access logs (default: ${DEFAULT-VALUE})")
   private final Boolean accessLogsEnabled = false;
 
+  @Option(
+      names = "--Xkey-store-parallel-processing-enabled",
+      description = "Set to false to disable parallel processing of key stores.",
+      paramLabel = "<BOOL>",
+      arity = "1",
+      hidden = true)
+  private boolean keystoreParallelProcessingEnabled = true;
+
   @CommandLine.Mixin private PicoCliTlsServerOptions picoCliTlsServerOptions;
 
   @Override
@@ -282,6 +290,11 @@ public Boolean isAccessLogsEnabled() {
     return accessLogsEnabled;
   }
 
+  @Override
+  public boolean keystoreParallelProcessingEnabled() {
+    return keystoreParallelProcessingEnabled;
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)

core/src/integrationTest/java/tech/pegasys/web3signer/core/jsonrpcproxy/support/TestBaseConfig.java

@@ -125,4 +125,9 @@ public Boolean isSwaggerUIEnabled() {
   public Boolean isAccessLogsEnabled() {
     return false;
   }
+
+  @Override
+  public boolean keystoreParallelProcessingEnabled() {
+    return true;
+  }
 }

core/src/main/java/tech/pegasys/web3signer/core/Eth1Runner.java

@@ -148,7 +148,7 @@ protected ArtifactSignerProvider createArtifactSignerProvider(
                     EthSecpArtifactSigner::new,
                     true);
 
-            return new SignerLoader()
+            return new SignerLoader(baseConfig.keystoreParallelProcessingEnabled())
                 .load(
                     baseConfig.getKeyConfigPath(),
                     "yaml",

core/src/main/java/tech/pegasys/web3signer/core/Eth2Runner.java

@@ -259,7 +259,7 @@ protected ArtifactSignerProvider createArtifactSignerProvider(
                         new BlsArtifactSigner(args.getKeyPair(), args.getOrigin(), args.getPath()));
 
             final MappedResults<ArtifactSigner> results =
-                new SignerLoader()
+                new SignerLoader(baseConfig.keystoreParallelProcessingEnabled())
                     .load(
                         baseConfig.getKeyConfigPath(),
                         "yaml",

core/src/main/java/tech/pegasys/web3signer/core/FilecoinRunner.java

@@ -132,7 +132,7 @@ protected ArtifactSignerProvider createArtifactSignerProvider(
                     signer -> new FcSecpArtifactSigner(signer, network),
                     false);
 
-            return new SignerLoader()
+            return new SignerLoader(baseConfig.keystoreParallelProcessingEnabled())
                 .load(
                     baseConfig.getKeyConfigPath(),
                     "yaml",

core/src/main/java/tech/pegasys/web3signer/core/config/BaseConfig.java

@@ -58,4 +58,6 @@ public interface BaseConfig {
   Boolean isSwaggerUIEnabled();
 
   Boolean isAccessLogsEnabled();
+
+  boolean keystoreParallelProcessingEnabled();
 }

signing/src/main/java/tech/pegasys/web3signer/signing/config/SignerLoader.java

@@ -14,7 +14,6 @@
 
 import tech.pegasys.signers.common.MappedResults;
 import tech.pegasys.web3signer.signing.ArtifactSigner;
-import tech.pegasys.web3signer.signing.config.metadata.HashicorpSigningMetadata;
 import tech.pegasys.web3signer.signing.config.metadata.SigningMetadata;
 import tech.pegasys.web3signer.signing.config.metadata.parser.SignerParser;
 
@@ -49,9 +48,19 @@ public class SignerLoader {
   private static final Logger LOG = LogManager.getLogger();
   private static final long FILES_PROCESSED_TO_REPORT = 10;
   private static final int MAX_FORK_JOIN_THREADS = 5;
+  // enable or disable parallel streams to convert and load private keys from metadata files
+  private final boolean useParallelStreams;
 
   private static final Map<Path, FileTime> metadataConfigFilesPathCache = new HashMap<>();
 
+  public SignerLoader(final boolean useParallelStreams) {
+    this.useParallelStreams = useParallelStreams;
+  }
+
+  public SignerLoader() {
+    this(true);
+  }
+
   public MappedResults<ArtifactSigner> load(
       final Path configsDirectory, final String fileExtension, final SignerParser signerParser) {
     final Instant start = Instant.now();
@@ -66,40 +75,30 @@ public MappedResults<ArtifactSigner> load(
         calculateTimeTaken(start));
 
     final Instant conversionStartInstant = Instant.now();
-    // convert yaml file content to list of SigningMetadata
+    // Step 1: convert yaml file content to list of SigningMetadata
     final MappedResults<SigningMetadata> signingMetadataResults =
         convertConfigFileContent(configFileContent.getContentMap(), signerParser);
 
     LOG.debug(
         "Signing configuration metadata files converted to signing metadata {}",
         signingMetadataResults.getValues().size());
 
-    final Collection<SigningMetadata> signingMetadata = signingMetadataResults.getValues();
-    // filter hashicorp signing metadata, we want them to be processed sequentially.
-    final Map<Boolean, List<SigningMetadata>> partitionedMetadata =
-        signingMetadata.stream()
-            .collect(
-                Collectors.partitioningBy(
-                    metadata -> !metadata.getType().equals(HashicorpSigningMetadata.TYPE)));
-    final MappedResults<ArtifactSigner> metadataResultParallel =
-        mapToArtifactSignerInParallel(partitionedMetadata.get(true), signerParser);
-    final MappedResults<ArtifactSigner> metadataResultSequential =
-        mapToArtifactSignerSequentially(partitionedMetadata.get(false), signerParser);
-
-    final MappedResults<ArtifactSigner> metadataResult =
-        MappedResults.merge(metadataResultParallel, metadataResultSequential);
+    // Step 2: Convert SigningMetadata to ArtifactSigners. This involves connecting to remote
+    // Hashicorp vault, AWS, Azure or decrypting local keystore files.
+    final MappedResults<ArtifactSigner> artifactSigners =
+        mapMetadataToArtifactSigner(signingMetadataResults.getValues(), signerParser);
 
     // merge error counts of config file parsing errors ...
-    metadataResult.mergeErrorCount(signingMetadataResults.getErrorCount());
-    metadataResult.mergeErrorCount(configFileContent.getErrorCount());
+    artifactSigners.mergeErrorCount(signingMetadataResults.getErrorCount());
+    artifactSigners.mergeErrorCount(configFileContent.getErrorCount());
 
     LOG.info(
         "Total Artifact Signer loaded via configuration files: {}\nError count {}\nTime Taken: {}.",
-        metadataResult.getValues().size(),
-        metadataResult.getErrorCount(),
+        artifactSigners.getValues().size(),
+        artifactSigners.getErrorCount(),
         calculateTimeTaken(conversionStartInstant));
 
-    return metadataResult;
+    return artifactSigners;
   }
 
   private MappedResults<SigningMetadata> convertConfigFileContent(
@@ -184,27 +183,32 @@ private SimpleEntry<Path, String> getMetadataFileContent(final Path path) throws
     return new SimpleEntry<>(path, Files.readString(path, StandardCharsets.UTF_8));
   }
 
-  private MappedResults<ArtifactSigner> mapToArtifactSignerInParallel(
+  private MappedResults<ArtifactSigner> mapMetadataToArtifactSigner(
       final Collection<SigningMetadata> signingMetadataCollection,
       final SignerParser signerParser) {
 
     if (signingMetadataCollection.isEmpty()) {
       return MappedResults.newSetInstance();
     }
 
-    LOG.info("Converting signing metadata to Artifact Signer using parallel streams ...");
+    LOG.info(
+        "Converting signing metadata to Artifact Signer using {} streams ...",
+        useParallelStreams ? "parallel" : "sequential");
 
     // use custom fork-join pool instead of common. Limit number of threads to avoid Azure bug
     ForkJoinPool forkJoinPool = null;
     try {
-      forkJoinPool = new ForkJoinPool(numberOfThreads());
-      return forkJoinPool
-          .submit(
-              () -> mapToArtifactSigner(signingMetadataCollection.parallelStream(), signerParser))
-          .get();
+      if (useParallelStreams) {
+        forkJoinPool = new ForkJoinPool(numberOfThreads());
+        return forkJoinPool
+            .submit(
+                () -> mapToArtifactSigner(signingMetadataCollection.parallelStream(), signerParser))
+            .get();
+      } else {
+        return mapToArtifactSigner(signingMetadataCollection.stream(), signerParser);
+      }
     } catch (final Exception e) {
-      LOG.error(
-          "Unexpected error in processing configuration files in parallel: {}", e.getMessage(), e);
+      LOG.error("Unexpected error in processing configuration files: {}", e.getMessage(), e);
       return MappedResults.errorResult();
     } finally {
       if (forkJoinPool != null) {
@@ -213,22 +217,6 @@ private MappedResults<ArtifactSigner> mapToArtifactSignerInParallel(
     }
   }
 
-  private MappedResults<ArtifactSigner> mapToArtifactSignerSequentially(
-      final Collection<SigningMetadata> signingMetadataCollection,
-      final SignerParser signerParser) {
-    if (signingMetadataCollection.isEmpty()) {
-      return MappedResults.newSetInstance();
-    }
-
-    LOG.info("Converting signing metadata to Artifact Signer using sequential streams ...");
-
-    try {
-      return mapToArtifactSigner(signingMetadataCollection.stream(), signerParser);
-    } catch (final Exception e) {
-      return MappedResults.errorResult();
-    }
-  }
-
   private MappedResults<ArtifactSigner> mapToArtifactSigner(
       final Stream<SigningMetadata> signingMetadataStream, final SignerParser signerParser) {
     final AtomicLong configFilesHandled = new AtomicLong();