Merge pull request #616 from volodymyr-babak/feature/audit-elastic

Added elastic sink. audit log service

@@ -317,3 +317,19 @@ audit_log:
       "user": "${AUDIT_LOG_MASK_USER:W}"
       "rule": "${AUDIT_LOG_MASK_RULE:W}"
       "plugin": "${AUDIT_LOG_MASK_PLUGIN:W}"
+  sink:
+    # Type of external sink. possible options: none, elasticsearch
+    type: "${AUDIT_LOG_SINK_TYPE:none}"
+    # Name of the index where audit logs stored
+    # Index name could contain next placeholders (not mandatory):
+    # @{TENANT} - substituted by tenant ID
+    # @{DATE} - substituted by current date in format provided in audit_log.sink.date_format
+    index_pattern: "${AUDIT_LOG_SINK_INDEX_PATTERN:@{TENANT}_AUDIT_LOG_@{DATE}}"
+    # Date format. Details of the pattern could be found here:
+    # https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html
+    date_format: "${AUDIT_LOG_SINK_DATE_FORMAT:YYYY.MM.DD}"
+    scheme_name: "${AUDIT_LOG_SINK_SCHEME_NAME:http}" # http or https
+    host: "${AUDIT_LOG_SINK_HOST:localhost}"
+    port: "${AUDIT_LOG_SINK_POST:9200}"
+    user_name: "${AUDIT_LOG_SINK_USER_NAME:}"
+    password: "${AUDIT_LOG_SINK_PASSWORD:}"
\ No newline at end of file

@@ -103,7 +103,12 @@
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-tx</artifactId>
-		</dependency>       
+		</dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-web</artifactId>
+            <scope>provided</scope>
+        </dependency>
 		<dependency>
 			<groupId>com.datastax.cassandra</groupId>
 			<artifactId>cassandra-driver-core</artifactId>
@@ -190,6 +195,10 @@
             <groupId>redis.clients</groupId>
             <artifactId>jedis</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>rest</artifactId>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

@@ -39,6 +39,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.page.TimePageData;
 import org.thingsboard.server.common.data.page.TimePageLink;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
+import org.thingsboard.server.dao.audit.sink.AuditLogSink;
 import org.thingsboard.server.dao.entity.EntityService;
 import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.dao.service.DataValidator;
@@ -69,6 +70,9 @@ public class AuditLogServiceImpl implements AuditLogService {
     @Autowired
     private EntityService entityService;
 
+    @Autowired
+    private AuditLogSink auditLogSink;
+
     @Override
     public TimePageData<AuditLog> findAuditLogsByTenantIdAndCustomerId(TenantId tenantId, CustomerId customerId, TimePageLink pageLink) {
         log.trace("Executing findAuditLogsByTenantIdAndCustomerId [{}], [{}], [{}]", tenantId, customerId, pageLink);
@@ -295,6 +299,9 @@ public class AuditLogServiceImpl implements AuditLogService {
         futures.add(auditLogDao.saveByTenantIdAndEntityId(auditLogEntry));
         futures.add(auditLogDao.saveByTenantIdAndCustomerId(auditLogEntry));
         futures.add(auditLogDao.saveByTenantIdAndUserId(auditLogEntry));
+
+        auditLogSink.logAction(auditLogEntry);
+
         return Futures.allAsList(futures);
     }
 

@@ -15,23 +15,20 @@
  */
 package org.thingsboard.server.dao.audit;
 
-import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.BaseData;
 import org.thingsboard.server.common.data.HasName;
-import org.thingsboard.server.common.data.User;
-import org.thingsboard.server.common.data.audit.ActionStatus;
 import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.audit.AuditLog;
 import org.thingsboard.server.common.data.id.*;
 import org.thingsboard.server.common.data.page.TimePageData;
 import org.thingsboard.server.common.data.page.TimePageLink;
 
-import java.util.Collections;
 import java.util.List;
 
+@Service
 @ConditionalOnProperty(prefix = "audit_log", value = "enabled", havingValue = "false")
 public class DummyAuditLogServiceImpl implements AuditLogService {
 

+/**
+ * Copyright © 2016-2017 The Thingsboard Authors
+ *
+ * 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 org.thingsboard.server.dao.audit.sink;
+
+import org.thingsboard.server.common.data.audit.AuditLog;
+
+public interface AuditLogSink {
+
+    void logAction(AuditLog auditLogEntry);
+}

+/**
+ * Copyright © 2016-2017 The Thingsboard Authors
+ *
+ * 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 org.thingsboard.server.dao.audit.sink;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.audit.AuditLog;
+
+@Component
+@ConditionalOnProperty(prefix = "audit_log.sink", value = "type", havingValue = "none")
+public class DummyAuditLogSink implements AuditLogSink {
+
+    @Override
+    public void logAction(AuditLog auditLogEntry) {
+    }
+}

+/**
+ * Copyright © 2016-2017 The Thingsboard Authors
+ *
+ * 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 org.thingsboard.server.dao.audit.sink;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.nio.entity.NStringEntity;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseListener;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.http.HttpMethod;
+import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.audit.AuditLog;
+import org.thingsboard.server.common.data.id.TenantId;
+
+import javax.annotation.PostConstruct;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+
+@Component
+@ConditionalOnProperty(prefix = "audit_log.sink", value = "type", havingValue = "elasticsearch")
+@Slf4j
+public class ElasticsearchAuditLogSink implements AuditLogSink {
+
+    private static final String TENANT_PLACEHOLDER = "@{TENANT}";
+    private static final String DATE_PLACEHOLDER = "@{DATE}";
+    private static final String INDEX_TYPE = "audit_log";
+
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @Value("${audit_log.sink.index_pattern}")
+    private String indexPattern;
+    @Value("${audit_log.sink.scheme_name}")
+    private String schemeName;
+    @Value("${audit_log.sink.host}")
+    private String host;
+    @Value("${audit_log.sink.port}")
+    private int port;
+    @Value("${audit_log.sink.user_name}")
+    private String userName;
+    @Value("${audit_log.sink.password}")
+    private String password;
+    @Value("${audit_log.sink.date_format}")
+    private String dateFormat;
+
+    private RestClient restClient;
+
+    @PostConstruct
+    public void init() {
+        try {
+            log.trace("Adding elastic rest endpoint... host [{}], port [{}], scheme name [{}]",
+                    host, port, schemeName);
+            RestClientBuilder builder = RestClient.builder(
+                    new HttpHost(host, port, schemeName));
+
+            if (StringUtils.isNotEmpty(userName) &&
+                    StringUtils.isNotEmpty(password)) {
+                log.trace("...using username [{}] and password ***", userName);
+                final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+                credentialsProvider.setCredentials(AuthScope.ANY,
+                        new UsernamePasswordCredentials(userName, password));
+                builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
+            }
+
+            this.restClient = builder.build();
+        } catch (Exception e) {
+            log.error("Sink init failed!", e);
+            throw new RuntimeException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void logAction(AuditLog auditLogEntry) {
+        String jsonContent = createElasticJsonRecord(auditLogEntry);
+
+        HttpEntity entity = new NStringEntity(
+                jsonContent,
+                ContentType.APPLICATION_JSON);
+
+        restClient.performRequestAsync(
+                HttpMethod.POST.name(),
+                String.format("/%s/%s", getIndexName(auditLogEntry.getTenantId()), INDEX_TYPE),
+                Collections.emptyMap(),
+                entity,
+                responseListener);
+    }
+
+    private String createElasticJsonRecord(AuditLog auditLog) {
+        ObjectNode auditLogNode = mapper.createObjectNode();
+        auditLogNode.put("postDate", LocalDateTime.now().toString());
+        auditLogNode.put("id", auditLog.getId().getId().toString());
+        auditLogNode.put("entityName", auditLog.getEntityName());
+        auditLogNode.put("tenantId", auditLog.getTenantId().getId().toString());
+        if (auditLog.getCustomerId() != null) {
+            auditLogNode.put("customerId", auditLog.getCustomerId().getId().toString());
+        }
+        auditLogNode.put("entityId", auditLog.getEntityId().getId().toString());
+        auditLogNode.put("entityType", auditLog.getEntityId().getEntityType().name());
+        auditLogNode.put("userId", auditLog.getUserId().getId().toString());
+        auditLogNode.put("userName", auditLog.getUserName());
+        auditLogNode.put("actionType", auditLog.getActionType().name());
+        if (auditLog.getActionData() != null) {
+            auditLogNode.put("actionData", auditLog.getActionData().toString());
+        }
+        auditLogNode.put("actionStatus", auditLog.getActionStatus().name());
+        auditLogNode.put("actionFailureDetails", auditLog.getActionFailureDetails());
+        return auditLogNode.toString();
+    }
+
+    private ResponseListener responseListener = new ResponseListener() {
+        @Override
+        public void onSuccess(Response response) {
+            log.trace("Elasticsearch sink log action method succeeded. Response result [{}]!", response);
+        }
+
+        @Override
+        public void onFailure(Exception exception) {
+            log.warn("Elasticsearch sink log action method failed!", exception);
+        }
+    };
+
+    private String getIndexName(TenantId tenantId) {
+        String indexName = indexPattern;
+        if (indexName.contains(TENANT_PLACEHOLDER) && tenantId != null) {
+            indexName = indexName.replace(TENANT_PLACEHOLDER, tenantId.getId().toString());
+        }
+        if (indexName.contains(DATE_PLACEHOLDER)) {
+            LocalDateTime now = LocalDateTime.now();
+            DateTimeFormatter formatter = DateTimeFormatter.ofPattern(dateFormat);
+            indexName = indexName.replace(DATE_PLACEHOLDER, now.format(formatter));
+        }
+        return indexName.toLowerCase();
+    }
+}

@@ -7,6 +7,7 @@ updates.enabled=false
 audit_log.enabled=true
 audit_log.by_tenant_partitioning=MONTHS
 audit_log.default_query_period=30
+audit_log.sink.type=none
 
 cache.type=caffeine
 #cache.type=redis

@@ -80,6 +80,7 @@
         <spring-test-dbunit.version>1.2.1</spring-test-dbunit.version>
         <postgresql.driver.version>9.4.1211</postgresql.driver.version>
         <sonar.exclusions>org/thingsboard/server/gen/**/*, org/thingsboard/server/extensions/core/plugin/telemetry/gen/**/*</sonar.exclusions>
+        <elasticsearch.version>5.0.2</elasticsearch.version>
     </properties>
 
     <modules>
@@ -803,6 +804,11 @@
                 <type>exe</type>
                 <scope>provided</scope>
             </dependency>
+            <dependency>
+                <groupId>org.elasticsearch.client</groupId>
+                <artifactId>rest</artifactId>
+                <version>${elasticsearch.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>