CE-124 oap-storage: run full sync when restarting the master

oap-storage/oap-storage-mongo/src/main/java/oap/storage/MongoPersistence.java

@@ -150,7 +150,7 @@ public void fsync() {
             var list = new ArrayList<WriteModel<Metadata<T>>>( batchSize );
             var deletedIds = new ArrayList<I>( batchSize );
             AtomicInteger updated = new AtomicInteger();
-            TransactionLog.ReplicationResult<I, Metadata<T>> updatedSince = storage.updatedSince( lastTimestamp );
+            TransactionLog.ReplicationResult<I, Metadata<T>> updatedSince = storage.updatedSince( lastTimestamp, hash );
 
             updatedSince.data.forEach( t -> {
                 updated.incrementAndGet();
@@ -167,6 +167,7 @@ public void fsync() {
             log.trace( "fsyncing, last: {}, updated objects in storage: {}, total in storage: {}", lastTimestamp, updated.get(), storage.size() );
             persist( deletedIds, list );
             lastTimestamp = updatedSince.timestamp;
+            hash = updatedSince.hash;
         } );
     }
 

oap-storage/oap-storage/src/main/java/oap/storage/AbstractPersistance.java

@@ -65,6 +65,7 @@ public abstract class AbstractPersistance<I, T> implements Closeable, AutoClosea
     protected ScheduledExecutorService scheduler;
     protected int batchSize = 100;
     protected volatile long lastTimestamp = -1;
+    protected volatile long hash = -1;
     protected volatile boolean stopped = false;
 
     public AbstractPersistance( MemoryStorage<I, T> storage, String tableName, long delay, Path crashDumpPath ) {

oap-storage/oap-storage/src/main/java/oap/storage/MemoryStorage.java

@@ -327,10 +327,10 @@ public void forEach( Consumer<? super Data> action ) {
     }
 
     @Override
-    public TransactionLog.ReplicationResult<Id, Metadata<Data>> updatedSince( long timestamp ) {
+    public TransactionLog.ReplicationResult<Id, Metadata<Data>> updatedSince( long timestamp, long hash ) {
         log.trace( "requested updated objects timestamp {}", timestamp );
 
-        return transactionLog.updatedSince( timestamp, memory.data.entrySet() );
+        return transactionLog.updatedSince( timestamp, hash, memory.data.entrySet() );
     }
 
     protected static class Memory<T, I> {

oap-storage/oap-storage/src/main/java/oap/storage/ReplicationMaster.java

@@ -25,5 +25,5 @@
 package oap.storage;
 
 public interface ReplicationMaster<Id, Data> {
-    TransactionLog.ReplicationResult<Id, Metadata<Data>> updatedSince( long timestamp );
+    TransactionLog.ReplicationResult<Id, Metadata<Data>> updatedSince( long timestamp, long hash );
 }

oap-storage/oap-storage/src/main/java/oap/storage/Replicator.java

@@ -33,14 +33,19 @@
 import oap.io.Closeables;
 import oap.storage.Storage.DataListener.IdObject;
 import oap.util.Cuid;
+import oap.util.Pair;
 
 import java.io.Closeable;
 import java.io.UncheckedIOException;
 import java.net.SocketException;
 import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static oap.storage.Storage.DataListener.IdObject.__io;
+import static oap.storage.TransactionLog.ReplicationResult.ReplicationStatusType.CHANGES;
 import static oap.storage.TransactionLog.ReplicationResult.ReplicationStatusType.FULL_SYNC;
+import static oap.util.Pair.__;
 
 /**
  * Replicator works on the MemoryStorage internals. It's intentional.
@@ -49,26 +54,42 @@
  */
 @Slf4j
 public class Replicator<I, T> implements Closeable {
+    final AtomicLong replicatorCounterFullSync = new AtomicLong();
+    final AtomicLong replicatorCounterPartialSync = new AtomicLong();
+    final AtomicLong replicatorSizeFullSync = new AtomicLong();
+    final AtomicLong replicatorSizePartialSync = new AtomicLong();
+
     private final MemoryStorage<I, T> slave;
     private final ReplicationMaster<I, T> master;
+    final ReentrantLock lock = new ReentrantLock();
     private String uniqueName = Cuid.UNIQUE.next();
-    private Scheduled scheduled;
+    private final Scheduled scheduled;
     private transient long timestamp = -1L;
+    private transient long hash = -1L;
 
     public Replicator( MemoryStorage<I, T> slave, ReplicationMaster<I, T> master, long interval ) {
-
         Preconditions.checkArgument( slave.transactionLog instanceof TransactionLogZero );
 
         this.slave = slave;
         this.master = master;
+
         this.scheduled = Scheduler.scheduleWithFixedDelay( getClass(), interval, i -> {
-            long newTimestamp = replicate( timestamp );
+            Pair<Long, Long> newTimestamp = replicate( timestamp );
             log.trace( "[{}] newTimestamp {}, lastModified {}", uniqueName, newTimestamp, timestamp );
 
-            timestamp = newTimestamp;
+            timestamp = newTimestamp._1;
+            hash = newTimestamp._2;
         } );
     }
 
+    public void start() {
+        Metrics.gauge( "replicator", Tags.of( "name", uniqueName, "type", FULL_SYNC.name() ), this, _ -> replicatorCounterFullSync.doubleValue() );
+        Metrics.gauge( "replicator", Tags.of( "name", uniqueName, "type", CHANGES.name() ), this, _ -> replicatorCounterPartialSync.doubleValue() );
+
+        Metrics.gauge( "replicator_size", Tags.of( "name", uniqueName, "type", FULL_SYNC.name() ), this, _ -> replicatorSizeFullSync.doubleValue() );
+        Metrics.gauge( "replicator_size", Tags.of( "name", uniqueName, "type", CHANGES.name() ), this, _ -> replicatorSizePartialSync.doubleValue() );
+    }
+
     public void replicateNow() {
         log.trace( "[{}] forcing replication...", uniqueName );
         scheduled.triggerNow();
@@ -79,72 +100,86 @@ public void replicateAllNow() {
         replicateNow();
     }
 
-    public synchronized long replicate( long timestamp ) {
-        log.trace( "replicate service {} timestamp {}", uniqueName, timestamp );
-
+    public Pair<Long, Long> replicate( long timestamp ) {
+        lock.lock();
         try {
-            log.trace( "[{}] replicate {} to {} timestamp: {}", master, slave, timestamp, uniqueName );
-            TransactionLog.ReplicationResult<I, Metadata<T>> updatedSince = master.updatedSince( timestamp );
-            log.trace( "[{}] type {} updated objects {}", uniqueName, updatedSince.type, updatedSince.data.size() );
+            log.trace( "replicate service {} timestamp {} hash {}", uniqueName, timestamp, hash );
 
-            Metrics.counter( "replicator", Tags.of( "name", uniqueName, "type", updatedSince.type.name() ) ).increment();
-            Metrics.counter( "replicator_size", Tags.of( "name", uniqueName, "type", updatedSince.type.name() ) ).increment( updatedSince.data.size() );
+            try {
+                log.trace( "[{}] replicate {} to {} timestamp {} hash {}", uniqueName, master, slave, timestamp, hash );
+                TransactionLog.ReplicationResult<I, Metadata<T>> updatedSince = master.updatedSince( timestamp, hash );
+                log.trace( "[{}] type {} updated objects {}", uniqueName, updatedSince.type, updatedSince.data.size() );
 
-            ArrayList<IdObject<I, T>> added = new ArrayList<>();
-            ArrayList<IdObject<I, T>> updated = new ArrayList<>();
-            ArrayList<IdObject<I, T>> deleted = new ArrayList<>();
+                switch( updatedSince.type ) {
+                    case FULL_SYNC -> {
+                        replicatorCounterFullSync.incrementAndGet();
+                        replicatorSizeFullSync.addAndGet( updatedSince.data.size() );
+                    }
+                    case CHANGES -> {
+                        replicatorCounterPartialSync.incrementAndGet();
+                        replicatorSizePartialSync.addAndGet( updatedSince.data.size() );
+                    }
+                    default -> throw new IllegalStateException( "Unknown replicator type: " + updatedSince.type );
+                }
 
-            long lastUpdate = updatedSince.timestamp;
+                ArrayList<IdObject<I, T>> added = new ArrayList<>();
+                ArrayList<IdObject<I, T>> updated = new ArrayList<>();
+                ArrayList<IdObject<I, T>> deleted = new ArrayList<>();
 
-            if( updatedSince.type == FULL_SYNC ) {
-                slave.memory.removePermanently();
-            }
+                long lastUpdate = updatedSince.timestamp;
 
-            for( TransactionLog.Transaction<I, Metadata<T>> transaction : updatedSince.data ) {
-                log.trace( "[{}] replicate {}", transaction, uniqueName );
+                if( updatedSince.type == FULL_SYNC ) {
+                    slave.memory.removePermanently();
+                }
 
-                Metadata<T> metadata = transaction.object;
-                I id = transaction.id;
+                for( TransactionLog.Transaction<I, Metadata<T>> transaction : updatedSince.data ) {
+                    log.trace( "[{}] replicate {}", transaction, uniqueName );
 
-                switch( transaction.operation ) {
-                    case INSERT -> {
-                        added.add( __io( id, metadata ) );
-                        slave.memory.put( id, metadata );
-                    }
-                    case UPDATE -> {
-                        if( metadata.isDeleted() ) {
-                            deleted.add( __io( id, metadata ) );
-                            slave.memory.removePermanently( id );
-                        } else {
-                            if( updatedSince.type == FULL_SYNC ) {
-                                added.add( __io( id, metadata ) );
+                    Metadata<T> metadata = transaction.object;
+                    I id = transaction.id;
+
+                    switch( transaction.operation ) {
+                        case INSERT -> {
+                            added.add( __io( id, metadata ) );
+                            slave.memory.put( id, metadata );
+                        }
+                        case UPDATE -> {
+                            if( metadata.isDeleted() ) {
+                                deleted.add( __io( id, metadata ) );
+                                slave.memory.removePermanently( id );
                             } else {
-                                updated.add( __io( id, metadata ) );
+                                if( updatedSince.type == FULL_SYNC ) {
+                                    added.add( __io( id, metadata ) );
+                                } else {
+                                    updated.add( __io( id, metadata ) );
+                                }
+                                slave.memory.put( id, metadata );
                             }
-                            slave.memory.put( id, metadata );
                         }
-                    }
-                    case DELETE -> {
-                        deleted.add( __io( id, metadata ) );
-                        slave.memory.removePermanently( id );
+                        case DELETE -> {
+                            deleted.add( __io( id, metadata ) );
+                            slave.memory.removePermanently( id );
+                        }
                     }
                 }
-            }
 
-            if( !added.isEmpty() || !updated.isEmpty() || !deleted.isEmpty() ) {
-                slave.fireChanged( added, updated, deleted );
-            }
+                if( !added.isEmpty() || !updated.isEmpty() || !deleted.isEmpty() ) {
+                    slave.fireChanged( added, updated, deleted );
+                }
 
-            return lastUpdate;
-        } catch( UncheckedIOException e ) {
-            log.error( e.getCause().getMessage() );
-            return timestamp;
-        } catch( Exception e ) {
-            if( e.getCause() instanceof SocketException ) {
+                return __( lastUpdate, updatedSince.hash );
+            } catch( UncheckedIOException e ) {
                 log.error( e.getCause().getMessage() );
-                return timestamp;
+                return __( timestamp, hash );
+            } catch( Exception e ) {
+                if( e.getCause() instanceof SocketException ) {
+                    log.error( e.getCause().getMessage() );
+                    return __( timestamp, hash );
+                }
+                throw e;
             }
-            throw e;
+        } finally {
+            lock.unlock();
         }
     }
 

oap-storage/oap-storage/src/main/java/oap/storage/TransactionLog.java

@@ -15,7 +15,7 @@ public interface TransactionLog<Id, T> {
 
     void delete( Id id, Metadata<T> data );
 
-    ReplicationResult<Id, Metadata<T>> updatedSince( long timestamp, Set<Map.Entry<Id, Metadata<T>>> fullData );
+    ReplicationResult<Id, Metadata<T>> updatedSince( long timestamp, long hash, Set<Map.Entry<Id, Metadata<T>>> fullData );
 
     enum Operation {
         INSERT,
@@ -45,11 +45,13 @@ class ReplicationResult<Id, T> implements Serializable {
         private static final long serialVersionUID = 467235422368462526L;
 
         public final long timestamp;
+        public final long hash;
         public final ReplicationStatusType type;
         public final List<Transaction<Id, T>> data;
 
-        public ReplicationResult( long timestamp, ReplicationStatusType type, List<Transaction<Id, T>> data ) {
+        public ReplicationResult( long timestamp, long hash, ReplicationStatusType type, List<Transaction<Id, T>> data ) {
             this.timestamp = timestamp;
+            this.hash = hash;
             this.type = type;
             this.data = data;
         }

oap-storage/oap-storage/src/main/java/oap/storage/TransactionLogImpl.java

@@ -2,6 +2,7 @@
 
 import oap.util.Lists;
 import org.apache.commons.collections4.queue.CircularFifoQueue;
+import org.joda.time.DateTimeUtils;
 import org.jspecify.annotations.NonNull;
 
 import java.util.ArrayList;
@@ -13,6 +14,7 @@
 public class TransactionLogImpl<Id, T> implements TransactionLog<Id, T> {
     public final CircularFifoQueue<Transaction<Id, Metadata<T>>> transactions;
     public final AtomicLong timestamp = new AtomicLong();
+    public long hash = DateTimeUtils.currentTimeMillis();
 
     public TransactionLogImpl( int transactionLogSize ) {
         this.transactions = new CircularFifoQueue<>( transactionLogSize );
@@ -34,15 +36,15 @@ public synchronized void delete( Id id, Metadata<T> data ) {
     }
 
     @Override
-    public synchronized ReplicationResult<Id, Metadata<T>> updatedSince( long timestamp, Set<Map.Entry<Id, Metadata<T>>> fullData ) {
+    public synchronized ReplicationResult<Id, Metadata<T>> updatedSince( long timestamp, long hash, Set<Map.Entry<Id, Metadata<T>>> fullData ) {
         int size = transactions.size();
-        if( size == 0 && timestamp < 0 ) { // first sync && no modification
+        if( this.hash != hash || timestamp < 0 ) { // first sync && no modification
             return fullSync( fullData, this.timestamp.longValue() );
         }
 
         Transaction<Id, Metadata<T>> older = transactions.peek();
         if( older == null ) {
-            return new ReplicationResult<>( this.timestamp.longValue(), ReplicationResult.ReplicationStatusType.CHANGES, List.of() );
+            return new ReplicationResult<>( this.timestamp.longValue(), this.hash, ReplicationResult.ReplicationStatusType.CHANGES, List.of() );
         }
         if( older.timestamp > timestamp ) {
             return fullSync( fullData, this.timestamp.longValue() );
@@ -56,10 +58,20 @@ public synchronized ReplicationResult<Id, Metadata<T>> updatedSince( long timest
             }
         }
 
-        return new ReplicationResult<>( this.timestamp.longValue(), ReplicationResult.ReplicationStatusType.CHANGES, list );
+        return new ReplicationResult<>( this.timestamp.longValue(), this.hash, ReplicationResult.ReplicationStatusType.CHANGES, list );
     }
 
     private @NonNull ReplicationResult<Id, Metadata<T>> fullSync( Set<Map.Entry<Id, Metadata<T>>> fullData, long t ) {
-        return new ReplicationResult<>( this.timestamp.longValue(), ReplicationResult.ReplicationStatusType.FULL_SYNC, Lists.map( fullData, d -> new Transaction<>( t, Operation.UPDATE, d.getKey(), d.getValue() ) ) );
+        return new ReplicationResult<>(
+            this.timestamp.longValue(),
+            this.hash,
+            ReplicationResult.ReplicationStatusType.FULL_SYNC,
+            Lists.map( fullData, d -> new Transaction<>( t, Operation.UPDATE, d.getKey(), d.getValue() ) ) );
+    }
+
+    public void reset() {
+        hash = DateTimeUtils.currentTimeMillis();
+        timestamp.set( 0 );
+        transactions.clear();
     }
 }

oap-storage/oap-storage/src/main/java/oap/storage/TransactionLogZero.java

@@ -23,11 +23,11 @@ public void delete( Id id, Metadata<T> data ) {
     }
 
     @Override
-    public ReplicationResult<Id, Metadata<T>> updatedSince( long timestamp, Set<Map.Entry<Id, Metadata<T>>> fullData ) {
+    public ReplicationResult<Id, Metadata<T>> updatedSince( long timestamp, long hash, Set<Map.Entry<Id, Metadata<T>>> fullData ) {
         throw new IllegalStateException();
     }
 
     private @NonNull ReplicationResult<Id, Metadata<T>> fullSync( Set<Map.Entry<Id, Metadata<T>>> fullData ) {
-        return new ReplicationResult<>( -1, ReplicationResult.ReplicationStatusType.FULL_SYNC, Lists.map( fullData, d -> new Transaction<>( 0L, Operation.UPDATE, d.getKey(), d.getValue() ) ) );
+        return new ReplicationResult<>( -1, 0L, ReplicationResult.ReplicationStatusType.FULL_SYNC, Lists.map( fullData, d -> new Transaction<>( 0L, Operation.UPDATE, d.getKey(), d.getValue() ) ) );
     }
 }

oap-storage/oap-storage/src/test/java/oap/storage/ReplicatorTest.java

@@ -28,6 +28,7 @@
 import oap.json.TypeIdFactory;
 import oap.testng.Fixtures;
 import oap.testng.SystemTimerFixture;
+import oap.util.Dates;
 import org.joda.time.DateTimeUtils;
 import org.testng.annotations.Test;
 
@@ -172,4 +173,43 @@ public void changed( List<IdObject<String, Bean>> added, List<IdObject<String, B
         }
     }
 
+    @Test
+    public void testFullResyncIfMasterRestarted() {
+        Dates.setTimeFixed( 1 );
+
+        MemoryStorage<String, Bean> slave = new MemoryStorage<>( Identifier.<Bean>forId( b -> b.id ).build(), SERIALIZED );
+        MemoryStorage<String, Bean> master = new MemoryStorage<>( Identifier.<Bean>forId( b -> b.id ).build(), SERIALIZED, 100 );
+        try( Replicator<String, Bean> replicator = new Replicator<>( slave, master, 100000 ) ) {
+            master.store( new Bean( "1" ), Storage.MODIFIED_BY_SYSTEM );
+
+            replicator.replicateNow();
+            assertEventually( 100, 100, () -> {
+                assertThat( replicator.replicatorSizeFullSync ).hasValue( 1L );
+                assertThat( replicator.replicatorSizePartialSync ).hasValue( 0L );
+            } );
+
+            Dates.incFixed( 10 );
+            replicator.replicateNow();
+            replicator.replicateNow();
+            assertEventually( 100, 100, () -> {
+                assertThat( replicator.replicatorCounterFullSync ).hasValue( 1L );
+                assertThat( replicator.replicatorCounterPartialSync ).hasValue( 2L );
+            } );
+
+
+            // restart
+            Dates.incFixed( 10 );
+            ( ( TransactionLogImpl<String, Bean> ) master.transactionLog ).reset();
+
+            master.store( new Bean( "2" ), Storage.MODIFIED_BY_SYSTEM );
+
+            replicator.replicateNow();
+            assertEventually( 100, 100, () -> {
+                assertThat( replicator.replicatorCounterFullSync ).hasValue( 2L );
+                assertThat( replicator.replicatorCounterPartialSync ).hasValue( 2L );
+            } );
+        }
+
+    }
+
 }