diff --git a/distro/pom.xml b/distro/pom.xml index 874b944f2..66f710896 100644 --- a/distro/pom.xml +++ b/distro/pom.xml @@ -136,6 +136,7 @@ atlas.graph.storage.hbase.regions-per-server=1 src/main/assemblies/atlas-repair-index-package.xml src/main/assemblies/classification-updater.xml + src/main/assemblies/notification-analyzer.xml apache-atlas-${project.version} gnu diff --git a/distro/src/main/assemblies/notification-analyzer.xml b/distro/src/main/assemblies/notification-analyzer.xml new file mode 100644 index 000000000..63d9f2490 --- /dev/null +++ b/distro/src/main/assemblies/notification-analyzer.xml @@ -0,0 +1,71 @@ + + + notification-analyzer + + zip + + + notification-analyzer + + + + + README* + + + + ../tools/notification-analyzer/target/dependency + . + + + ../tools/notification-analyzer/scripts + . + + *.sh + + 0755 + 0755 + + + ../tools/notification-analyzer/src/main/resources + . + + atlas-log4j.xml + atlas-application.properties + + + + ../tools/notification-analyzer + . + + README + + + + ../tools/notification-analyzer/target + . + + atlas-notification-analyzer-${project.version}.jar + + + + diff --git a/pom.xml b/pom.xml index 2f9aedbd2..bfb1094ff 100644 --- a/pom.xml +++ b/pom.xml @@ -846,6 +846,7 @@ addons/kafka-bridge tools/classification-updater tools/atlas-index-repair + tools/notification-analyzer addons/impala-hook-api addons/impala-bridge-shim addons/impala-bridge diff --git a/tools/notification-analyzer/README b/tools/notification-analyzer/README new file mode 100644 index 000000000..7631e06d9 --- /dev/null +++ b/tools/notification-analyzer/README @@ -0,0 +1,119 @@ +# +# 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. +# + +Introduction + This utility analyzes hook notification messages stored in JSON format + in a file, and reports following details: + - number of notifications per notification type + - number of entities created/updated + - number of entity references in notifications per entity type + - number of entity operations performed + - number of entity operations performed per entity type + +Setup + - All libraries necessary to run the utility are packaged in following file: + distro/target/apache-atlas--notification-analyzer.zip + + - Unzip the file in the directory where the tool needs to be installed. + + - Update log configurations in atlas-log4j.xml + + +Running the utility + - Execute following command to start the utility: + ./notification-analyzer.sh -m message_file.json [-o output_file] + + - Progress will be printed in the following format in the output file (if specified) or on stdout: + PROGRESS #1: analyzed 1000 notifications, 1071 messages + PROGRESS #2: analyzed 2000 notifications, 2131 messages + PROGRESS #3: analyzed 3000 notifications, 3194 messages + ... + Completed analyzing 114755 notification, 120816 messages. Time taken: 234 seconds + + - Note that the number of notifications might be less than the number + of messages in the file in case some notifications were split into + multiple messages due to their size. + + - Logs will be printed in file /tmp/atlas-notification-analyzer.log. The location of + the log file can be configured using following environment variables: + LOGFILE_DIR + LOGFILE_NAME + +Sample Result: + The utility will print the result of analysis in the following format: + { + "notifications": 114755, + "notificationLengthAvg": 74331, + "notificationLengthMax": 101148684, + "splitNotifications": 453, + "splitNotificationLengthAvg": 13901446, + "splitNotificationLengthMax": 101148684, + "entities": 598435, + "notificationEntities": 2575347, + "notificationByType": { + "ENTITY_CREATE_V2": 49428, + "ENTITY_FULL_UPDATE_V2": 1597, + "ENTITY_PARTIAL_UPDATE_V2": 36561, + "ENTITY_DELETE_V2": 27169 + }, + "notificationEntityByType": { + "hdfs_path": 16417, + "hive_db": 20471, + "hive_table": 57143, + "hive_storagedesc": 30018, + "hive_column": 685384, + "hive_process": 41512 + "hive_column_lineage": 1724402, + }, + "entityOperations": { + "CREATE": 598435, + "UPDATE": 1913182 + "PARTIAL_UPDATE": 36561, + "DELETE": 27169 + }, + "entityOperationsByType": { + "CREATE": { + "hdfs_path": 10940, + "hive_db": 224, + "hive_table": 22154, + "hive_storagedesc": 15280, + "hive_column": 332332, + "hive_process": 23462, + "hive_column_lineage": 194043 + }, + "UPDATE" { + "hdfs_path": 5477, + "hive_column": 319559, + "hive_column_lineage": 1530359, + "hive_db": 20203, + "hive_process": 18050, + "hive_storagedesc": 13204, + "hive_table": 6330 + }, + "PARTIAL_UPDATE": { + "hive_column": 33493, + "hive_storagedesc": 1534, + "hive_table": 1534 + }, + "DELETE": { + "hive_db": 44, + "hive_table": 27125 + } + } + } + diff --git a/tools/notification-analyzer/pom.xml b/tools/notification-analyzer/pom.xml new file mode 100644 index 000000000..2f772a76f --- /dev/null +++ b/tools/notification-analyzer/pom.xml @@ -0,0 +1,192 @@ + + + + + 4.0.0 + + apache-atlas + org.apache.atlas + 3.0.0-SNAPSHOT + ../../ + + atlas-notification-analyzer + Apache Atlas Notification Analyzer + Apache Atlas Notification Analyzer + jar + + + + org.slf4j + slf4j-api + ${slf4j.version} + + + + org.slf4j + slf4j-log4j12 + ${slf4j.version} + + + + org.apache.atlas + atlas-notification + ${project.version} + + + commons-cli + commons-cli + ${commons-cli.version} + + + commons-collections + commons-collections + ${commons-collections.version} + + + + + + dist + + + + org.apache.maven.plugins + maven-dependency-plugin + + + copy-binaries + package + + copy + + + ${project.build.directory}/dependency + false + false + true + + + ${project.groupId} + ${project.artifactId} + ${project.version} + + + ${project.groupId} + atlas-intg + ${project.version} + + + ${project.groupId} + atlas-notification + ${project.version} + + + commons-cli + commons-cli + ${commons-cli.version} + + + commons-codec + commons-codec + ${commons-codec.version} + + + commons-collections + commons-collections + ${commons-collections.version} + + + commons-io + commons-io + ${commons-io.version} + + + org.apache.commons + commons-lang3 + + + org.apache.commons + commons-compress + 1.26.2 + + + commons-configuration + commons-configuration + ${commons-conf.version} + + + commons-lang + commons-lang + ${commons-lang.version} + + + commons-logging + commons-logging + ${commons-logging.version} + + + com.fasterxml.jackson.core + jackson-annotations + ${jackson.version} + + + com.fasterxml.jackson.core + jackson-core + ${jackson.version} + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.databind.version} + + + log4j + log4j + ${log4j.version} + + + org.apache.logging.log4j + log4j-core + ${log4j2.version} + + + org.apache.logging.log4j + log4j-api + ${log4j2.version} + + + org.slf4j + slf4j-api + ${slf4j.version} + + + org.slf4j + slf4j-log4j12 + ${slf4j.version} + + + + + + + + + + + diff --git a/tools/notification-analyzer/scripts/notification-analyzer.sh b/tools/notification-analyzer/scripts/notification-analyzer.sh new file mode 100644 index 000000000..179336bd0 --- /dev/null +++ b/tools/notification-analyzer/scripts/notification-analyzer.sh @@ -0,0 +1,28 @@ +#!/bin/bash +# +# 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. See accompanying LICENSE file. +# +# resolve links - $0 may be a softlink + +M2_REPO=~/.m2/repository + +LIB_DIR=$(pwd) +LOGFILE_DIR="${LOGFILE_DIR:-/tmp/}" +LOGFILE_NAME="${LOGFILE_NAME:-atlas-notification-analyzer.log}" + +CP=. +for i in "${LIB_DIR}/"*.jar; do + CP="${CP}:$i" +done + +java -cp ${CP} -Dlog4j.configuration=atlas-log4j.xml -Datlas.log.dir=${LOGFILE_DIR} -Datlas.log.file=${LOGFILE_NAME} org.apache.atlas.tools.NotificationAnalyzer $* diff --git a/tools/notification-analyzer/src/main/java/org/apache/atlas/tools/NotificationAnalyzer.java b/tools/notification-analyzer/src/main/java/org/apache/atlas/tools/NotificationAnalyzer.java new file mode 100644 index 000000000..20b106546 --- /dev/null +++ b/tools/notification-analyzer/src/main/java/org/apache/atlas/tools/NotificationAnalyzer.java @@ -0,0 +1,322 @@ +/** + * 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 org.apache.atlas.tools; + +import org.apache.atlas.model.instance.AtlasEntity; +import org.apache.atlas.model.instance.AtlasObjectId; +import org.apache.atlas.model.notification.HookNotification; +import org.apache.atlas.notification.AtlasNotificationMessageDeserializer; +import org.apache.atlas.notification.NotificationInterface.NotificationType; +import org.apache.atlas.utils.AtlasJson; +import org.apache.atlas.v1.model.instance.Referenceable; +import org.apache.atlas.v1.model.notification.HookNotificationV1.EntityCreateRequest; +import org.apache.atlas.v1.model.notification.HookNotificationV1.EntityDeleteRequest; +import org.apache.atlas.v1.model.notification.HookNotificationV1.EntityPartialUpdateRequest; +import org.apache.atlas.v1.model.notification.HookNotificationV1.EntityUpdateRequest; +import org.apache.atlas.model.notification.HookNotification.EntityCreateRequestV2; +import org.apache.atlas.model.notification.HookNotification.EntityDeleteRequestV2; +import org.apache.atlas.model.notification.HookNotification.EntityPartialUpdateRequestV2; +import org.apache.atlas.model.notification.HookNotification.EntityUpdateRequestV2; +import org.apache.commons.cli.BasicParser; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.Options; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.util.*; +import java.util.concurrent.atomic.AtomicInteger; + +public class NotificationAnalyzer { + private static final Logger LOG = LoggerFactory.getLogger(NotificationAnalyzer.class); + + private final String msgFile; + private final String outputFile; + private final AtlasNotificationMessageDeserializer deserializer; + private final Map notificationCountByType = new HashMap<>(); + private final AtomicInteger entityCount = new AtomicInteger(); + private final Map entityCountByType = new HashMap<>(); + private final Map entityOperCount = new HashMap<>(); + private final Map> entityOperByTypeCount = new HashMap<>(); + private final Set knownEntities = new HashSet<>(); + private final IntSummaryStatistics notificationStats = new IntSummaryStatistics(); + private final IntSummaryStatistics splitNotificationStats = new IntSummaryStatistics(); + + public static void main(String[] args) { + CommandLineParser parser = new BasicParser(); + Options options = new Options(); + + options.addOption("m", "message-file", true, "Messages file"); + options.addOption("o", "output-file", true, "Output file"); + + try { + CommandLine cmd = parser.parse(options, args); + String msgFile = cmd.getOptionValue("m"); + String outFile = cmd.getOptionValue("o"); + + if (msgFile == null || msgFile.isEmpty()) { + msgFile = "ATLAS_HOOK.json"; + } + + NotificationAnalyzer analyzer = new NotificationAnalyzer(msgFile, outFile, NotificationType.HOOK); + + analyzer.analyze(); + } catch (Throwable t) { + throw new RuntimeException(t); + } + } + + public NotificationAnalyzer(String msgFile, String outputFile, NotificationType notificationType) { + this.msgFile = msgFile; + this.outputFile = outputFile; + this.deserializer = notificationType.getDeserializer(); + } + + public void analyze() throws Exception { + long startTimeMs = System.currentTimeMillis(); + + try (BufferedReader reader = getInputReader(); PrintWriter writer = getOutputWriter()) { + int msgCount = 0; + int notificationSize = 0; + + for (String msg = reader.readLine(); msg != null; msg = reader.readLine()) { + msgCount++; + notificationSize += msg.length(); + + HookNotification notification = (HookNotification) deserializer.deserialize(msg); + + if (notification == null) { // split notification, continue + continue; + } + + notificationStats.accept(notificationSize); + + if (notificationSize > msg.length()) { + splitNotificationStats.accept(notificationSize); + } + + notificationSize = 0; + + notificationCountByType.computeIfAbsent(notification.getType().name(), e -> new AtomicInteger()).incrementAndGet(); + + switch (notification.getType()) { + case ENTITY_CREATE: + handleEntityCreate((EntityCreateRequest) notification); + break; + case ENTITY_PARTIAL_UPDATE: + handleEntityPartialUpdate((EntityPartialUpdateRequest) notification); + break; + case ENTITY_FULL_UPDATE: + handleEntityUpdate((EntityUpdateRequest) notification); + break; + case ENTITY_DELETE: + handleEntityDelete((EntityDeleteRequest) notification); + break; + case ENTITY_CREATE_V2: + handleEntityCreateV2((EntityCreateRequestV2) notification); + break; + case ENTITY_PARTIAL_UPDATE_V2: + handleEntityPartialUpdateV2((EntityPartialUpdateRequestV2) notification); + break; + case ENTITY_FULL_UPDATE_V2: + handleEntityUpdateV2((EntityUpdateRequestV2) notification); + break; + case ENTITY_DELETE_V2: + handleEntityDeleteV2((EntityDeleteRequestV2) notification); + break; + } + + if ((notificationStats.getCount() % 1000) == 0) { + LOG.info("PROGRESS #{}: analyzed {} notifications, {} messages", (msgCount / 1000), notificationStats.getCount(), msgCount); + writer.printf("PROGRESS #%1$d: analyzed %2$d notifications, %3$d messages%n", (msgCount / 1000), notificationStats.getCount(), msgCount); + writer.flush(); + } + } + + long timeTakenSeconds = (System.currentTimeMillis() - startTimeMs) / 1000; + + LOG.info("Completed analyzing {}. Time taken: {} seconds", msgFile, timeTakenSeconds); + writer.printf("Completed analyzing %1$s. Time taken: %2$d seconds%n", msgFile, timeTakenSeconds); + writer.flush(); + + Map results = new LinkedHashMap<>(); + + results.put("messages", msgCount); + results.put("notifications", notificationStats.getCount()); + results.put("notificationLengthAvg", (int) notificationStats.getAverage()); + results.put("notificationLengthMax", notificationStats.getMax()); + results.put("splitNotifications", splitNotificationStats.getCount()); + results.put("splitNotificationLengthAvg", (int) splitNotificationStats.getAverage()); + results.put("splitNotificationLengthMax", splitNotificationStats.getMax()); + results.put("entities", entityCount); + results.put("notificationEntities", entityCountByType.values().stream().mapToInt(AtomicInteger::get).sum()); + results.put("notificationByType", notificationCountByType); + results.put("notificationEntityByType", entityCountByType); + results.put("entityOperations", entityOperCount); + results.put("entityOperationsByType", entityOperByTypeCount); + + String msg = AtlasJson.toJson(results); + + LOG.info(msg); + writer.println(msg); + writer.flush(); + } + } + + private void handleEntityCreate(EntityCreateRequest request) { + if (request.getEntities() != null) { + for (Referenceable entity : request.getEntities()) { + recordEntity(entity); + } + } + } + + private void handleEntityUpdate(EntityUpdateRequest request) { + if (request.getEntities() != null) { + for (Referenceable entity : request.getEntities()) { + recordEntity(entity); + } + } + } + + private void handleEntityPartialUpdate(EntityPartialUpdateRequest request) { + recordEntityOperation(request.getTypeName(), "PARTIAL_UPDATE"); + } + + private void handleEntityDelete(EntityDeleteRequest request) { + knownEntities.remove(getEntityKey(request)); + + recordEntityOperation(request.getTypeName(), "DELETE"); + } + + private void handleEntityCreateV2(EntityCreateRequestV2 request) { + if (request.getEntities() != null) { + if (request.getEntities().getEntities() != null) { + for (AtlasEntity entity : request.getEntities().getEntities()) { + recordEntity(entity); + } + } + + if (request.getEntities().getReferredEntities() != null) { + for (AtlasEntity entity : request.getEntities().getReferredEntities().values()) { + recordEntity(entity); + } + } + } + } + + private void handleEntityUpdateV2(EntityUpdateRequestV2 request) { + if (request.getEntities() != null) { + if (request.getEntities().getEntities() != null) { + for (AtlasEntity entity : request.getEntities().getEntities()) { + recordEntity(entity); + } + } + + if (request.getEntities().getReferredEntities() != null) { + for (AtlasEntity entity : request.getEntities().getReferredEntities().values()) { + recordEntity(entity); + } + } + } + } + + private void handleEntityPartialUpdateV2(EntityPartialUpdateRequestV2 request) { + if (request.getEntity() != null && request.getEntity().getEntity() != null) { + AtlasEntity entity = request.getEntity().getEntity(); + + recordEntityOperation(entity.getTypeName(), "PARTIAL_UPDATE"); + } + } + + private void handleEntityDeleteV2(EntityDeleteRequestV2 request) { + if (request.getEntities() != null) { + for (AtlasObjectId objId : request.getEntities()) { + knownEntities.remove(getEntityKey(objId)); + + recordEntityOperation(objId.getTypeName(), "DELETE"); + } + } + } + + private void recordEntity(AtlasEntity entity) { + final String operation; + + if (knownEntities.add(getEntityKey(entity))) { + operation = "CREATE"; + } else { + operation = "UPDATE"; + } + + recordEntityOperation(entity.getTypeName(), operation); + } + + private void recordEntity(Referenceable entity) { + final String operation; + + if (knownEntities.add(getEntityKey(entity))) { + operation = "CREATE"; + } else { + operation = "UPDATE"; + } + + recordEntityOperation(entity.getTypeName(), operation); + } + + private void recordEntityOperation(String entityTypeName, String operation) { + if (operation.equals("CREATE")) { + entityCount.incrementAndGet(); + } + + entityCountByType.computeIfAbsent(entityTypeName, c -> new AtomicInteger()).incrementAndGet(); + entityOperCount.computeIfAbsent(operation, c -> new AtomicInteger()).incrementAndGet(); + entityOperByTypeCount.computeIfAbsent(operation, c -> new TreeMap<>()).computeIfAbsent(entityTypeName, c -> new AtomicInteger()).incrementAndGet(); + } + + private String getEntityKey(AtlasEntity entity) { + return entity.getTypeName() + ":" + getUniqueKey(entity.getAttributes()); + } + + private String getEntityKey(AtlasObjectId objectId) { + return objectId.getTypeName() + ":" + getUniqueKey(objectId.getUniqueAttributes()); + } + + private String getEntityKey(Referenceable entity) { + return entity.getTypeName() + ":" + getUniqueKey(entity.getValues()); + } + + private String getEntityKey(EntityDeleteRequest request) { + return request.getTypeName() + ":" + request.getAttributeValue(); + } + + private Object getUniqueKey(Map attributes) { + Object ret = attributes.get("qualifiedName"); + + return ret == null ? attributes.get("name") : ret; + } + + private BufferedReader getInputReader() throws IOException { + return new BufferedReader(new FileReader(msgFile)); + } + + private PrintWriter getOutputWriter() throws IOException { + return (outputFile == null || outputFile.isEmpty()) ? new PrintWriter(System.out) : new PrintWriter(new FileWriter(outputFile, true)); + } +} diff --git a/tools/notification-analyzer/src/main/resources/atlas-application.properties b/tools/notification-analyzer/src/main/resources/atlas-application.properties new file mode 100644 index 000000000..01d705720 --- /dev/null +++ b/tools/notification-analyzer/src/main/resources/atlas-application.properties @@ -0,0 +1,18 @@ +# +# 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. +# + diff --git a/tools/notification-analyzer/src/main/resources/atlas-log4j.xml b/tools/notification-analyzer/src/main/resources/atlas-log4j.xml new file mode 100755 index 000000000..0f9182f36 --- /dev/null +++ b/tools/notification-analyzer/src/main/resources/atlas-log4j.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + +