fix(analytics): Fix SSL issue with analytics on frontend (datahub-pro…

…ject#2840)

datahub-frontend/app/react/analytics/AnalyticsService.java

@@ -42,8 +42,8 @@ public class AnalyticsService {
   private final Logger _logger = LoggerFactory.getLogger(AnalyticsService.class.getName());
 
   private final RestHighLevelClient _elasticClient;
+  private final Optional<String> _indexPrefix;
 
-  private static final String INDEX_NAME = "datahub_usage_event";
   private static final String FILTERED = "filtered";
   private static final String DATE_HISTOGRAM = "date_histogram";
   private static final String UNIQUE = "unique";
@@ -58,17 +58,24 @@ public class AnalyticsService {
   public static final String DATA_JOB_INDEX = "datajobindex_v2";
   public static final String DATASET_INDEX = "datasetindex_v2";
 
-  public AnalyticsService(final RestHighLevelClient elasticClient) {
+  public AnalyticsService(final RestHighLevelClient elasticClient, final Optional<String> indexPrefix) {
     _elasticClient = elasticClient;
+    _indexPrefix = indexPrefix;
+  }
+
+  private String getIndexName(String baseIndexName) {
+    return _indexPrefix.map(p -> p + "_").orElse("") + baseIndexName;
   }
 
   public List<NamedLine> getTimeseriesChart(String indexName, DateRange dateRange, DateInterval granularity,
       Optional<String> dimension, // Length 1 for now
       Map<String, List<String>> filters, Optional<String> uniqueOn) {
+
+    String finalIndexName = getIndexName(indexName);
     _logger.debug(
         String.format("Invoked getTimeseriesChart with indexName: %s, dateRange: %s, granularity: %s, dimension: %s,",
-            indexName, dateRange, granularity, dimension)
-        + String.format("filters: %s, uniqueOn: %s", filters, uniqueOn));
+            finalIndexName, dateRange, granularity, dimension) + String.format("filters: %s, uniqueOn: %s", filters,
+            uniqueOn));
 
     AggregationBuilder filteredAgg = getFilteredAggregation(filters, ImmutableMap.of(), Optional.of(dateRange));
 
@@ -84,7 +91,7 @@ public List<NamedLine> getTimeseriesChart(String indexName, DateRange dateRange,
       filteredAgg.subAggregation(dateHistogram);
     }
 
-    SearchRequest searchRequest = constructSearchRequest(indexName, filteredAgg);
+    SearchRequest searchRequest = constructSearchRequest(finalIndexName, filteredAgg);
     Aggregations aggregationResult = executeAndExtract(searchRequest).getAggregations();
     try {
       if (dimension.isPresent()) {
@@ -117,10 +124,10 @@ private List<NumericDataPoint> extractPointsFromAggregations(Aggregations aggreg
   public List<NamedBar> getBarChart(String indexName, Optional<DateRange> dateRange, List<String> dimensions,
       // Length 1 or 2
       Map<String, List<String>> filters, Optional<String> uniqueOn) {
+    String finalIndexName = getIndexName(indexName);
     _logger.debug(
-        String.format("Invoked getBarChart with indexName: %s, dateRange: %s, dimensions: %s,",
-            indexName, dateRange, dimensions)
-            + String.format("filters: %s, uniqueOn: %s", filters, uniqueOn));
+        String.format("Invoked getBarChart with indexName: %s, dateRange: %s, dimensions: %s,", finalIndexName,
+            dateRange, dimensions) + String.format("filters: %s, uniqueOn: %s", filters, uniqueOn));
 
     assert (dimensions.size() == 1 || dimensions.size() == 2);
     AggregationBuilder filteredAgg = getFilteredAggregation(filters, ImmutableMap.of(), dateRange);
@@ -136,7 +143,7 @@ public List<NamedBar> getBarChart(String indexName, Optional<DateRange> dateRang
     }
     filteredAgg.subAggregation(termAgg);
 
-    SearchRequest searchRequest = constructSearchRequest(indexName, filteredAgg);
+    SearchRequest searchRequest = constructSearchRequest(finalIndexName, filteredAgg);
     Aggregations aggregationResult = executeAndExtract(searchRequest).getAggregations();
 
     try {
@@ -169,10 +176,10 @@ private List<BarSegment> extractBarSegmentsFromAggregations(Aggregations aggrega
 
   public List<Row> getTopNTableChart(String indexName, Optional<DateRange> dateRange, String groupBy,
       Map<String, List<String>> filters, Optional<String> uniqueOn, int maxRows) {
+    String finalIndexName = getIndexName(indexName);
     _logger.debug(
-        String.format("Invoked getTopNTableChart with indexName: %s, dateRange: %s, groupBy: %s",
-            indexName, dateRange, groupBy)
-            + String.format("filters: %s, uniqueOn: %s", filters, uniqueOn));
+        String.format("Invoked getTopNTableChart with indexName: %s, dateRange: %s, groupBy: %s", finalIndexName,
+            dateRange, groupBy) + String.format("filters: %s, uniqueOn: %s", filters, uniqueOn));
 
     AggregationBuilder filteredAgg = getFilteredAggregation(filters, ImmutableMap.of(), dateRange);
 
@@ -183,7 +190,7 @@ public List<Row> getTopNTableChart(String indexName, Optional<DateRange> dateRan
     }
     filteredAgg.subAggregation(termAgg);
 
-    SearchRequest searchRequest = constructSearchRequest(indexName, filteredAgg);
+    SearchRequest searchRequest = constructSearchRequest(finalIndexName, filteredAgg);
     Aggregations aggregationResult = executeAndExtract(searchRequest).getAggregations();
 
     try {
@@ -200,10 +207,14 @@ public List<Row> getTopNTableChart(String indexName, Optional<DateRange> dateRan
 
   public int getHighlights(String indexName, Optional<DateRange> dateRange, Map<String, List<String>> filters,
       Map<String, List<String>> mustNotFilters, Optional<String> uniqueOn) {
+    String finalIndexName = getIndexName(indexName);
+    _logger.debug(String.format("Invoked getHighlights with indexName: %s, dateRange: %s", finalIndexName, dateRange)
+        + String.format("filters: %s, uniqueOn: %s", filters, uniqueOn));
+
     AggregationBuilder filteredAgg = getFilteredAggregation(filters, mustNotFilters, dateRange);
     uniqueOn.ifPresent(s -> filteredAgg.subAggregation(getUniqueQuery(s)));
 
-    SearchRequest searchRequest = constructSearchRequest(indexName, filteredAgg);
+    SearchRequest searchRequest = constructSearchRequest(finalIndexName, filteredAgg);
     Filter aggregationResult = executeAndExtract(searchRequest);
     try {
       if (uniqueOn.isPresent()) {

datahub-frontend/app/react/analytics/AnalyticsServiceModule.java

@@ -1,6 +1,7 @@
 package react.analytics;
 
 import com.google.inject.AbstractModule;
+import java.util.Optional;
 import org.apache.http.ssl.SSLContextBuilder;
 import play.Environment;
 import javax.annotation.Nonnull;
@@ -14,6 +15,8 @@
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
 import java.security.cert.CertificateException;
+import utils.ConfigUtil;
+
 
 /**
  * Guice module responsible for configuring & creating an instance of {@link AnalyticsService}.
@@ -33,6 +36,7 @@ public class AnalyticsServiceModule extends AbstractModule {
     private static final String ELASTIC_CLIENT_CONNECTION_REQUEST_TIMEOUT_PATH = "analytics.elastic.connectionRequestTimeout";
     private static final String ELASTIC_CLIENT_USERNAME_PATH = "analytics.elastic.username";
     private static final String ELASTIC_CLIENT_PASSWORD_PATH = "analytics.elastic.password";
+    private static final String ELASTIC_INDEX_PREFIX = "analytics.elastic.indexPrefix";
 
     /*
         Required SSL Config Paths
@@ -61,29 +65,32 @@ public AnalyticsServiceModule(final Environment environment, final com.typesafe.
     @Override
     protected void configure() {
         final SSLContext context = createSSLContext();
-        final AnalyticsService backend = new AnalyticsService(ElasticClientFactory.createElasticClient(
-                _configs.hasPath(ELASTIC_CLIENT_USE_SSL_PATH) && _configs.getBoolean(ELASTIC_CLIENT_USE_SSL_PATH),
+        final AnalyticsService backend = new AnalyticsService(
+            ElasticClientFactory.createElasticClient(
+                ConfigUtil.getBoolean(_configs, ELASTIC_CLIENT_USE_SSL_PATH),
                 _configs.getString(ELASTIC_CLIENT_HOST_PATH),
                 _configs.getInt(ELASTIC_CLIENT_PORT_PATH),
-                _configs.hasPath(ELASTIC_CLIENT_THREAD_COUNT_PATH) ? _configs.getInt(ELASTIC_CLIENT_THREAD_COUNT_PATH) : DEFAULT_THREAD_COUNT,
-                _configs.hasPath(ELASTIC_CLIENT_CONNECTION_REQUEST_TIMEOUT_PATH) ? _configs.getInt(ELASTIC_CLIENT_CONNECTION_REQUEST_TIMEOUT_PATH) : DEFAULT_CONNECTION_TIMEOUT,
-                _configs.hasPath(ELASTIC_CLIENT_USERNAME_PATH) ? _configs.getString(ELASTIC_CLIENT_USERNAME_PATH) : null,
-                _configs.hasPath(ELASTIC_CLIENT_PASSWORD_PATH) ? _configs.getString(ELASTIC_CLIENT_PASSWORD_PATH) : null,
+                ConfigUtil.getInt(_configs, ELASTIC_CLIENT_THREAD_COUNT_PATH, DEFAULT_THREAD_COUNT),
+                ConfigUtil.getInt(_configs, ELASTIC_CLIENT_CONNECTION_REQUEST_TIMEOUT_PATH, DEFAULT_CONNECTION_TIMEOUT),
+                ConfigUtil.getString(_configs, ELASTIC_CLIENT_USERNAME_PATH, null),
+                ConfigUtil.getString(_configs, ELASTIC_CLIENT_PASSWORD_PATH, null),
                 context
-        ));
+            ),
+            Optional.ofNullable(ConfigUtil.getString(_configs, ELASTIC_INDEX_PREFIX, null))
+        );
         bind(AnalyticsService.class).toInstance(backend);
     }
 
     private SSLContext createSSLContext() {
 
-        final String sslProtocol = _configs.hasPath(ELASTIC_CLIENT_SSL_PROTOCOL_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_PROTOCOL_PATH) : null;
-        final String sslTrustStoreFile = _configs.hasPath(ELASTIC_CLIENT_SSL_TRUST_STORE_FILE_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_TRUST_STORE_FILE_PATH) : null;
-        final String sslTrustStoreType = _configs.hasPath(ELASTIC_CLIENT_SSL_TRUST_STORE_TYPE_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_TRUST_STORE_TYPE_PATH) : null;
-        final String sslTrustStorePassword = _configs.hasPath(ELASTIC_CLIENT_SSL_TRUST_STORE_PASSWORD_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_TRUST_STORE_PASSWORD_PATH): null;
-        final String sslKeyStoreFile = _configs.hasPath(ELASTIC_CLIENT_SSL_KEY_STORE_FILE_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_KEY_STORE_FILE_PATH) : null;
-        final String sslKeyStoreType = _configs.hasPath(ELASTIC_CLIENT_SSL_KEY_STORE_TYPE_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_KEY_STORE_TYPE_PATH) : null;
-        final String sslKeyStorePassword = _configs.hasPath(ELASTIC_CLIENT_SSL_KEY_STORE_PASSWORD_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_KEY_STORE_PASSWORD_PATH) : null;
-        final String sslSecureRandomImplementation = _configs.hasPath(ELASTIC_CLIENT_SSL_SECURE_RANDOM_IMPL_PATH) ? _configs.getString(ELASTIC_CLIENT_SSL_SECURE_RANDOM_IMPL_PATH) : null;
+        final String sslProtocol = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_PROTOCOL_PATH, null);
+        final String sslTrustStoreFile = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_TRUST_STORE_FILE_PATH, null);
+        final String sslTrustStoreType = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_TRUST_STORE_TYPE_PATH, null);
+        final String sslTrustStorePassword = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_TRUST_STORE_PASSWORD_PATH, null);
+        final String sslKeyStoreFile = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_KEY_STORE_FILE_PATH, null);
+        final String sslKeyStoreType = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_KEY_STORE_TYPE_PATH, null);
+        final String sslKeyStorePassword = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_KEY_STORE_PASSWORD_PATH, null);
+        final String sslSecureRandomImplementation = ConfigUtil.getString(_configs, ELASTIC_CLIENT_SSL_SECURE_RANDOM_IMPL_PATH, null);
 
         final SSLContextBuilder sslContextBuilder = new SSLContextBuilder();
         if (sslProtocol != null) {

datahub-frontend/app/react/controllers/TrackingController.java

@@ -23,7 +23,10 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.Properties;
+import utils.ConfigUtil;
+
 
 public class TrackingController extends Controller {
 
@@ -86,6 +89,11 @@ protected void finalize() {
         _producer.close();
     }
 
+    private void setConfig(Properties props, String key, String configKey) {
+        Optional.ofNullable(ConfigUtil.getString(_config, configKey, null))
+            .ifPresent(v -> props.put(key, v));
+    }
+
     private KafkaProducer createKafkaProducer() {
         final Properties props = new Properties();
         props.put(ProducerConfig.CLIENT_ID_CONFIG, "datahub-frontend");
@@ -97,22 +105,22 @@ private KafkaProducer createKafkaProducer() {
         if (_config.hasPath(securityProtocolConfig)
                 && KAFKA_SSL_PROTOCOLS.contains(_config.getString(securityProtocolConfig))) {
             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, _config.getString(securityProtocolConfig));
-            props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, _config.getString("analytics.kafka.ssl.key.password"));
+            setConfig(props, SslConfigs.SSL_KEY_PASSWORD_CONFIG, "analytics.kafka.ssl.key.password");
 
-            props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, _config.getString("analytics.kafka.ssl.keystore.type"));
-            props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, _config.getString("analytics.kafka.ssl.keystore.location"));
-            props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, _config.getString("analytics.kafka.ssl.keystore.password"));
+            setConfig(props, SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "analytics.kafka.ssl.keystore.type");
+            setConfig(props, SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "analytics.kafka.ssl.keystore.location");
+            setConfig(props, SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "analytics.kafka.ssl.keystore.password");
 
-            props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, _config.getString("analytics.kafka.ssl.truststore.type"));
-            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, _config.getString("analytics.kafka.ssl.truststore.location"));
-            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, _config.getString("analytics.kafka.ssl.truststore.password"));
+            setConfig(props, SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "analytics.kafka.ssl.truststore.type");
+            setConfig(props, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "analytics.kafka.ssl.truststore.location");
+            setConfig(props, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "analytics.kafka.ssl.truststore.password");
 
-            props.put(SslConfigs.SSL_PROTOCOL_CONFIG, _config.getString("analytics.kafka.ssl.protocol"));
-            props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, _config.getString("analytics.kafka.ssl.endpoint.identification.algorithm"));
+            setConfig(props, SslConfigs.SSL_PROTOCOL_CONFIG, "analytics.kafka.ssl.protocol");
+            setConfig(props, SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "analytics.kafka.ssl.endpoint.identification.algorithm");
 
             if (_config.getString(securityProtocolConfig).equals(SecurityProtocol.SASL_SSL.name())) {
-                props.put(SaslConfigs.SASL_MECHANISM, _config.getString("analytics.kafka.sasl.mechanism"));
-                props.put(SaslConfigs.SASL_JAAS_CONFIG, _config.getString("analytics.kafka.sasl.jaas.config"));
+                setConfig(props, SaslConfigs.SASL_MECHANISM, "analytics.kafka.sasl.mechanism");
+                setConfig(props, SaslConfigs.SASL_JAAS_CONFIG, "analytics.kafka.sasl.jaas.config");
             }
         }
 

datahub-frontend/app/utils/ConfigUtil.java

@@ -0,0 +1,18 @@
+package utils;
+
+import com.typesafe.config.Config;
+
+
+public class ConfigUtil {
+  public static boolean getBoolean(Config config, String key) {
+    return config.hasPath(key) && config.getBoolean(key);
+  }
+
+  public static int getInt(Config config, String key, int defaultValue) {
+    return config.hasPath(key) ? config.getInt(key) : defaultValue;
+  }
+
+  public static String getString(Config config, String key, String defaultValue) {
+    return config.hasPath(key) ? config.getString(key) : defaultValue;
+  }
+}

datahub-frontend/conf/application.conf

@@ -184,4 +184,6 @@ analytics.elastic.sslContext.sslKeyStorePassword = ${?ELASTIC_CLIENT_SSL_KEY_STO
 
 # Optional username + password for use with SSL
 analytics.elastic.username = ${?ELASTIC_CLIENT_USERNAME}
-analytics.elastic.password = ${?ELASTIC_CLIENT_PASSWORD}
+analytics.elastic.password = ${?ELASTIC_CLIENT_PASSWORD}
+
+analytics.elastic.indexPrefix = ${?ELASTIC_INDEX_PREFIX}

datahub-kubernetes/datahub/charts/datahub-frontend/templates/deployment.yaml

@@ -26,14 +26,20 @@ spec:
       securityContext:
         {{- toYaml .Values.podSecurityContext | nindent 8 }}
       volumes:
+        {{- with .Values.global.credentialsAndCertsSecrets }}
+        - name: datahub-certs-dir
+          secret:
+            defaultMode: 0444
+            secretName: {{ .name }}
+        {{- end }}
+        {{- if .Values.exporters.jmx.enabled }}
+        - name: config-jmx-exporter
+          configMap:
+            name: {{ include "datahub-gms.fullname" . }}-config-jmx-exporter
+        {{- end }}
       {{- with .Values.extraVolumes }}
         {{- toYaml . | nindent 8 }}
       {{- end }}
-      {{- if .Values.exporters.jmx.enabled }}
-      - name: config-jmx-exporter
-        configMap:
-          name: {{ include "datahub-frontend.fullname" . }}-config-jmx-exporter
-      {{- end }}
       initContainers:
       {{- with .Values.extraInitContainers }}
         {{- toYaml . | nindent 8 }}
@@ -82,8 +88,26 @@ spec:
               value: "{{ .Values.global.datahub.appVersion }}"
             - name: DATAHUB_PLAY_MEM_BUFFER_SIZE
               value: "{{ .Values.datahub.play.mem.buffer.size }}"
+            - name: DATAHUB_ANALYTICS_ENABLED
+              value: "{{ .Values.global.datahub_analytics_enabled }}"
+            {{- if .Values.global.datahub_analytics_enabled }}
             - name: KAFKA_BOOTSTRAP_SERVER
               value: "{{ .Values.global.kafka.bootstrap.server }}"
+            {{- if .Values.global.springKafkaConfigurationOverrides }}
+            {{- range $configName, $configValue := .Values.global.springKafkaConfigurationOverrides }}
+            - name: KAFKA_PROPERTIES_{{ $configName | replace "." "_" | upper }}
+              value: {{ $configValue }}
+            {{- end }}
+            {{- end }}
+            {{- if .Values.global.credentialsAndCertsSecrets }}
+            {{- range $envVarName, $envVarValue := .Values.global.credentialsAndCertsSecrets.secureEnv }}
+            - name: KAFKA_PROPERTIES_{{ $envVarName | replace "." "_" | upper }}
+              valueFrom:
+                secretKeyRef:
+                  name: {{ $.Values.global.credentialsAndCertsSecrets.name }}
+                  key: {{ $envVarValue }}
+            {{- end }}
+            {{- end }}
             - name: ELASTIC_CLIENT_HOST
               value: "{{ .Values.global.elasticsearch.host }}"
             - name: ELASTIC_CLIENT_PORT
@@ -101,14 +125,22 @@ spec:
                   name: "{{ .password.secretRef }}"
                   key: "{{ .password.secretKey }}"
             {{- end }}
+            {{- if .Values.global.kafka.topics }}
+            - name: DATAHUB_TRACKING_TOPIC
+              value: {{ .Values.global.kafka.topics.datahub_usage_event_name}}
+            {{- else }}
             - name: DATAHUB_TRACKING_TOPIC
               value: "DataHubUsageEvent_v1"
-            - name: DATAHUB_ANALYTICS_ENABLED
-              value: "{{ .Values.global.datahub_analytics_enabled }}"
+            {{- end }}
+            {{- end }}
           {{- with .Values.extraEnvs }}
             {{- toYaml . | nindent 12 }}
           {{- end }}
           volumeMounts:
+          {{- with .Values.global.credentialsAndCertsSecrets }}
+            - name: datahub-certs-dir
+              mountPath: {{ .path | default "/mnt/certs" }}
+          {{- end }}
           {{- with .Values.extraVolumeMounts }}
             {{- toYaml . | nindent 12 }}
           {{- end }}