OAP-228 Move oap-statsdb into the oap project

oap-statsdb/oap-statsdb-master/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+    <name>oap-stats-db-master</name>
+    <artifactId>oap-stats-db-master</artifactId>
+
+    <parent>
+        <groupId>oap</groupId>
+        <artifactId>oap-statsdb-parent</artifactId>
+        <version>${oap.project.version}</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>oap</groupId>
+            <artifactId>oap-statsdb</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>oap</groupId>
+            <artifactId>oap-stdlib</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>oap</groupId>
+            <artifactId>oap-storage-mongo</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>oap</groupId>
+            <artifactId>oap-storage-mongo-test</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>${oap.deps.lombok.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>

oap-statsdb/oap-statsdb-master/src/main/java/oap/statsdb/JsonCodec.java

@@ -0,0 +1,52 @@
+package oap.statsdb;
+
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import lombok.SneakyThrows;
+import oap.json.Binder;
+import oap.reflect.TypeRef;
+import org.bson.BsonReader;
+import org.bson.BsonWriter;
+import org.bson.Document;
+import org.bson.codecs.Codec;
+import org.bson.codecs.DecoderContext;
+import org.bson.codecs.DocumentCodec;
+import org.bson.codecs.EncoderContext;
+
+/**
+ * Created by igor.petrenko on 26.03.2019.
+ */
+class JsonNodeCodec implements Codec<MongoNode> {
+    private final DocumentCodec documentCodec;
+    private ObjectWriter fileWriter;
+    private ObjectReader fileReader;
+
+    JsonNodeCodec() {
+        this.documentCodec = new DocumentCodec();
+        var ref = new TypeRef<MongoNode>() {
+        };
+        this.fileReader = Binder.json.readerFor( ref );
+        this.fileWriter = Binder.json.writerFor( ref );
+    }
+
+    @SneakyThrows
+    @Override
+    public MongoNode decode( BsonReader bsonReader, DecoderContext decoderContext ) {
+        var doc = documentCodec.decode( bsonReader, decoderContext );
+
+        return fileReader.readValue( Binder.json.marshal( doc ) );
+    }
+
+    @SneakyThrows
+    @Override
+    public void encode( BsonWriter bsonWriter, MongoNode data, EncoderContext encoderContext ) {
+        var doc = Document.parse( fileWriter.writeValueAsString( data ) );
+
+        documentCodec.encode( bsonWriter, doc, encoderContext );
+    }
+
+    @Override
+    public Class<MongoNode> getEncoderClass() {
+        return MongoNode.class;
+    }
+}

oap-statsdb/oap-statsdb-master/src/main/java/oap/statsdb/MongoNode.java

@@ -0,0 +1,25 @@
+package oap.statsdb;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.util.Map;
+
+/**
+ * Created by igor.petrenko on 26.03.2019.
+ */
+@ToString
+@EqualsAndHashCode( of = { "_id" } )
+public class MongoNode {
+    @SuppressWarnings( "checkstyle:MemberName" )
+    public final Map<String, String> _id;
+    public final Node n;
+
+    @JsonCreator
+    @SuppressWarnings( "checkstyle:ParameterName" )
+    public MongoNode( Map<String, String> _id, Node n ) {
+        this._id = _id;
+        this.n = n;
+    }
+}

oap-statsdb/oap-statsdb-master/src/main/java/oap/statsdb/StatsDBMaster.java

@@ -0,0 +1,161 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) Open Application Platform Authors
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package oap.statsdb;
+
+import lombok.extern.slf4j.Slf4j;
+import oap.statsdb.RemoteStatsDB.Sync;
+import oap.util.Lists;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+@Slf4j
+public class StatsDBMaster extends StatsDB implements Closeable, Runnable {
+    private final StatsDBStorage storage;
+
+    public StatsDBMaster( NodeSchema schema, StatsDBStorage storage ) {
+        super( schema );
+        this.storage = storage;
+
+        db.putAll( storage.load( schema ) );
+        init( db.values() );
+    }
+
+    private void merge( String key, Node masterNode, Node rNode, List<List<String>> retList, int level ) {
+        if( log.isTraceEnabled() )
+            log.trace( "merge {}/{}[{}]::{}", schema.get( level ).key, schema.get( level ).clazz, level, key );
+        assert Objects.equals( masterNode.v.getClass(), rNode.v.getClass() )
+            : "[" + level + "]/" + key + "::" + masterNode.v.getClass() + " vs " + rNode.v.getClass();
+
+        var list = merge( masterNode.db, rNode.db, retList, level );
+        list.forEach( l -> l.add( 0, key ) );
+
+        retList.addAll( list );
+
+        var ret = masterNode.merge( rNode );
+        if( !ret ) {
+            var k = new ArrayList<String>();
+            k.add( key );
+            retList.add( k );
+        }
+    }
+
+    private List<List<String>> merge( Map<String, Node> masterDB, Map<String, Node> remoteDB, List<List<String>> retList, int level ) {
+        for( var entry : remoteDB.entrySet() ) {
+            var key = entry.getKey();
+            var rNode = entry.getValue();
+
+            var masterNode = masterDB.computeIfAbsent( key, k -> new Node( schema.get( level + 1 ).newInstance() ) );
+
+            merge( key, masterNode, rNode, retList, level + 1 );
+        }
+
+        return retList;
+    }
+
+    private List<List<String>> merge( ArrayList<Sync.NodeIdNode> remoteDB ) {
+        assert remoteDB != null;
+
+        var retList = new ArrayList<List<String>>();
+
+        var remoteDbTree = toTree( remoteDB );
+
+        remoteDbTree.forEach( ( key, rnode ) -> {
+            var mnode = db.computeIfAbsent( key, k -> new Node( schema.get( 0 ).newInstance() ) );
+
+            merge( key, mnode, rnode, retList, 0 );
+            updateAggregates( mnode );
+        } );
+
+        return retList;
+    }
+
+    private Map<String, Node> toTree( ArrayList<Sync.NodeIdNode> remoteDB ) {
+        var ret = new HashMap<String, Node>();
+
+        for( var nodeIdNode : remoteDB ) {
+            var node = nodeIdNode.node;
+            var nodeId = nodeIdNode.nodeId;
+
+            Node treeNode = null;
+            for( var i = 0; i < nodeId.size(); i++ ) {
+                var key = nodeId.get( i );
+                var finalI = i;
+                treeNode = ( treeNode != null ? treeNode.db
+                    : ret ).computeIfAbsent( key, k -> new Node( schema.get( finalI ).newInstance() ) );
+            }
+
+            treeNode.set( node );
+        }
+
+        return ret;
+    }
+
+    @SuppressWarnings( "unchecked" )
+    private void init( Collection<Node> nodes ) {
+        nodes.forEach( node -> {
+            if( node.v instanceof Node.Container ) {
+                init( node.db.values() );
+                ( ( Node.Container ) node.v ).aggregate( Lists.map( node.db.values(), b -> b.v ) );
+            }
+        } );
+    }
+
+    public boolean update( Sync sync, String host ) {
+        assert sync != null;
+        assert sync.data != null;
+
+        synchronized( host.intern() ) {
+            var failedKeys = merge( sync.data );
+
+            if( !failedKeys.isEmpty() ) {
+                log.error( "failed keys:" );
+                failedKeys.forEach( key -> log.error( "[{}]: {}", host, key ) );
+            }
+
+            return true;
+        }
+    }
+
+    public void reset() {
+        removeAll();
+        storage.removeAll();
+    }
+
+    @Override
+    public void close() {
+        storage.store( schema, db );
+    }
+
+    @Override
+    public void run() {
+        storage.store( schema, db );
+    }
+}

oap-statsdb/oap-statsdb-master/src/main/java/oap/statsdb/StatsDBMessageListener.java

@@ -0,0 +1,43 @@
+package oap.statsdb;
+
+import lombok.extern.slf4j.Slf4j;
+import oap.json.Binder;
+import oap.message.MessageListener;
+import oap.message.MessageProtocol;
+
+import java.io.ByteArrayInputStream;
+
+import static oap.statsdb.StatsDBTransportMessage.MESSAGE_TYPE;
+
+/**
+ * Created by igor.petrenko on 2019-12-17.
+ */
+@Slf4j
+public class StatsDBMessageListener implements MessageListener {
+    private final StatsDBMaster master;
+
+    public StatsDBMessageListener( StatsDBMaster master ) {
+        this.master = master;
+    }
+
+    @Override
+    public byte getId() {
+        return MESSAGE_TYPE;
+    }
+
+    @Override
+    public String getInfo() {
+        return "stats-db";
+    }
+
+    @Override
+    public short run( int version, String hostName, int size, byte[] data, String md5 ) {
+        log.trace( "new stats version {} hostName {} size {} md5 {} data '{}'",
+            version, hostName, size, md5, new String( data ) );
+
+        var sync = Binder.json.unmarshal( RemoteStatsDB.Sync.class, new ByteArrayInputStream( data ) );
+        master.update( sync, hostName );
+
+        return MessageProtocol.STATUS_OK;
+    }
+}

oap-statsdb/oap-statsdb-master/src/main/java/oap/statsdb/StatsDBStorage.java

@@ -0,0 +1,40 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) Open Application Platform Authors
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package oap.statsdb;
+
+import java.util.Map;
+
+/**
+ * Created by igor.petrenko on 26.03.2019.
+ */
+public interface StatsDBStorage {
+    StatsDBStorage NULL = new StatsDBStorageNull();
+
+    Map<String, Node> load( NodeSchema schema );
+
+    void store( NodeSchema schema, Map<String, Node> db );
+
+    void removeAll();
+}