Add PostgreSQLConnectorConfig to expose SSL configuration options (#6181)

* Add PostgreSQLConnectorConfig to expose SSL configuration options for the Postgres Metadata Storage module.

* Fix checkstyle violations and add license header

* Convert properties in the postgres docs to be the full property path and fix typo

* Fix grammar in sslFactory docs

docs/content/development/extensions-core/postgresql.md

@@ -47,3 +47,19 @@ Make sure to [include](../../operations/including-extensions.html) `postgresql-m
   druid.metadata.storage.connector.user=druid
   druid.metadata.storage.connector.password=diurd
   ```
+
+## Configuration
+In most cases, the configuration options map directly to the [postgres jdbc connection options](https://jdbc.postgresql.org/documentation/head/connect.html).
+
+|Property|Description|Default|Required|
+|--------|-----------|-------|--------|
+| `druid.metadata.postgres.useSSL` | Enables SSL | `false` | no |
+| `druid.metadata.postgres.sslPassword` | The [Password Provider](../../operations/password-provider.html) or String password for the client's key. | none | no |
+| `druid.metadata.postgres.sslFactory` | The class name to use as the `SSLSocketFactory` | none | no |
+| `druid.metadata.postgres.sslFactoryArg` | An optional argument passed to the sslFactory's constructor | none | no |
+| `druid.metadata.postgres.sslMode` | The sslMode. Possible values are "disable", "require", "verify-ca", "verify-full", "allow" and "prefer"| none | no |
+| `druid.metadata.postgres.sslCert` | The full path to the certificate file. | none | no |
+| `druid.metadata.postgres.sslKey` | The full path to the key file. | none | no |
+| `druid.metadata.postgres.sslRootCert` | The full path to the root certificate. | none | no |
+| `druid.metadata.postgres.sslHostNameVerifier` | The classname of the hostname verifier. | none | no |
+| `druid.metadata.postgres.sslPasswordCallback` | The classname of the SSL password provider. | none | no |

extensions-core/postgresql-metadata-storage/src/main/java/io/druid/metadata/storage/postgresql/PostgreSQLConnector.java

@@ -27,6 +27,7 @@ import io.druid.metadata.MetadataStorageConnectorConfig;
 import io.druid.metadata.MetadataStorageTablesConfig;
 import io.druid.metadata.SQLMetadataConnector;
 import org.apache.commons.dbcp2.BasicDataSource;
+import org.postgresql.PGProperty;
 import org.skife.jdbi.v2.DBI;
 import org.skife.jdbi.v2.Handle;
 import org.skife.jdbi.v2.tweak.HandleCallback;
@@ -48,7 +49,11 @@ public class PostgreSQLConnector extends SQLMetadataConnector
   private volatile Boolean canUpsert;
 
   @Inject
-  public PostgreSQLConnector(Supplier<MetadataStorageConnectorConfig> config, Supplier<MetadataStorageTablesConfig> dbTables)
+  public PostgreSQLConnector(
+      Supplier<MetadataStorageConnectorConfig> config,
+      Supplier<MetadataStorageTablesConfig> dbTables,
+      PostgreSQLConnectorConfig connectorConfig
+  )
   {
     super(config, dbTables);
 
@@ -58,6 +63,46 @@ public class PostgreSQLConnector extends SQLMetadataConnector
     datasource.setDriverClassLoader(getClass().getClassLoader());
     datasource.setDriverClassName("org.postgresql.Driver");
 
+    // SSL Configuration
+    if (connectorConfig.isUseSSL()) {
+      log.info("SSL is enabled on this PostgreSQL connection.");
+      datasource.addConnectionProperty(PGProperty.SSL.getName(), String.valueOf(connectorConfig.isUseSSL()));
+
+      if (connectorConfig.getPassword() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_PASSWORD.getName(), connectorConfig.getPassword());
+      }
+      if (connectorConfig.getSslFactory() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_FACTORY.getName(), connectorConfig.getSslFactory());
+      }
+      if (connectorConfig.getSslFactoryArg() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_FACTORY_ARG.getName(), connectorConfig.getSslFactoryArg());
+      }
+      if (connectorConfig.getSslMode() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_MODE.getName(), connectorConfig.getSslMode());
+      }
+      if (connectorConfig.getSslCert() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_CERT.getName(), connectorConfig.getSslCert());
+      }
+      if (connectorConfig.getSslKey() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_KEY.getName(), connectorConfig.getSslKey());
+      }
+      if (connectorConfig.getSslRootCert() != null) {
+        datasource.addConnectionProperty(PGProperty.SSL_ROOT_CERT.getName(), connectorConfig.getSslRootCert());
+      }
+      if (connectorConfig.getSslHostNameVerifier() != null) {
+        datasource.addConnectionProperty(
+            PGProperty.SSL_HOSTNAME_VERIFIER.getName(),
+            connectorConfig.getSslHostNameVerifier()
+        );
+      }
+      if (connectorConfig.getSslPasswordCallback() != null) {
+        datasource.addConnectionProperty(
+            PGProperty.SSL_PASSWORD_CALLBACK.getName(),
+            connectorConfig.getSslPasswordCallback()
+        );
+      }
+    }
+
     this.dbi = new DBI(datasource);
 
     log.info("Configured PostgreSQL as metadata storage");

extensions-core/postgresql-metadata-storage/src/main/java/io/druid/metadata/storage/postgresql/PostgreSQLConnectorConfig.java

@@ -0,0 +1,123 @@
+/*
+ * 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 io.druid.metadata.storage.postgresql;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.druid.metadata.PasswordProvider;
+
+public class PostgreSQLConnectorConfig
+{
+  @JsonProperty
+  private boolean useSSL = false;
+
+  @JsonProperty("sslPassword")
+  private PasswordProvider sslPasswordProvider;
+
+  @JsonProperty
+  private String sslFactory;
+
+  @JsonProperty
+  private String sslFactoryArg;
+
+  @JsonProperty
+  private String sslMode;
+
+  @JsonProperty
+  private String sslCert;
+
+  @JsonProperty
+  private String sslKey;
+
+  @JsonProperty
+  private String sslRootCert;
+
+  @JsonProperty
+  private String sslHostNameVerifier;
+
+  @JsonProperty
+  private String sslPasswordCallback;
+
+
+  public boolean isUseSSL()
+  {
+    return useSSL;
+  }
+
+  public String getPassword()
+  {
+    return sslPasswordProvider == null ? null : sslPasswordProvider.getPassword();
+  }
+
+  public String getSslFactory()
+  {
+    return sslFactory;
+  }
+
+  public String getSslFactoryArg()
+  {
+    return sslFactoryArg;
+  }
+
+  public String getSslMode()
+  {
+    return sslMode;
+  }
+
+  public String getSslCert()
+  {
+    return sslCert;
+  }
+
+  public String getSslKey()
+  {
+    return sslKey;
+  }
+
+  public String getSslRootCert()
+  {
+    return sslRootCert;
+  }
+
+  public String getSslHostNameVerifier()
+  {
+    return sslHostNameVerifier;
+  }
+
+  public String getSslPasswordCallback()
+  {
+    return sslPasswordCallback;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "PostgreSQLConnectorConfig{" +
+           "useSSL='" + useSSL + '\'' +
+           ", sslFactory='" + sslFactory + '\'' +
+           ", sslFactoryArg='" + sslFactoryArg + '\'' +
+           ", sslMode='" + sslMode + '\'' +
+           ", sslCert='" + sslCert + '\'' +
+           ", sslKey='" + sslKey + '\'' +
+           ", sslRootCert='" + sslRootCert + '\'' +
+           ", sslHostNameVerifier='" + sslHostNameVerifier + '\'' +
+           ", sslPasswordCallback='" + sslPasswordCallback + '\'' +
+           '}';
+  }
+}

extensions-core/postgresql-metadata-storage/src/main/java/io/druid/metadata/storage/postgresql/PostgreSQLMetadataStorageModule.java

@@ -23,6 +23,7 @@ import com.fasterxml.jackson.databind.Module;
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Binder;
 import com.google.inject.Key;
+import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
 import io.druid.guice.PolyBind;
 import io.druid.guice.SQLMetadataStorageDruidModule;
@@ -57,6 +58,8 @@ public class PostgreSQLMetadataStorageModule extends SQLMetadataStorageDruidModu
   {
     super.configure(binder);
 
+    JsonConfigProvider.bind(binder, "druid.metadata.postgres.ssl", PostgreSQLConnector.class);
+
     PolyBind
         .optionBinder(binder, Key.get(MetadataStorageProvider.class))
         .addBinding(TYPE)

extensions-core/postgresql-metadata-storage/src/test/java/io/druid/metadata/storage/postgresql/PostgreSQLConnectorTest.java

@@ -49,7 +49,8 @@ public class PostgreSQLConnectorTest
                 null,
                 null
             )
-        )
+        ),
+        new PostgreSQLConnectorConfig()
     );
 
     Assert.assertTrue(connector.isTransientException(new SQLException("bummer, connection problem", "08DIE")));