diff --git a/oap-application/oap-application/src/main/java/oap/application/Kernel.java b/oap-application/oap-application/src/main/java/oap/application/Kernel.java index 7252551488..46f41ca9e9 100644 --- a/oap-application/oap-application/src/main/java/oap/application/Kernel.java +++ b/oap-application/oap-application/src/main/java/oap/application/Kernel.java @@ -479,7 +479,7 @@ private void startService( Supervisor supervisor, ModuleItem.ServiceItem si ) { Service service = si.service; Object instance = si.instance; if( service.supervision.supervise ) { - supervisor.startSupervised( si.serviceName, instance, + supervisor.startSupervised( si, instance, service.supervision.preStartWith, service.supervision.startWith, service.supervision.preStopWith, @@ -488,13 +488,13 @@ private void startService( Supervisor supervisor, ModuleItem.ServiceItem si ) { } if( service.supervision.thread ) { - supervisor.startThread( si.serviceName, instance, applicationConfiguration.shutdown ); + supervisor.startThread( si, instance, applicationConfiguration.shutdown ); } else { if( service.supervision.schedule && service.supervision.cron != null ) - supervisor.scheduleCron( si.serviceName, ( Runnable ) instance, + supervisor.scheduleCron( si, ( Runnable ) instance, service.supervision.cron ); else if( service.supervision.schedule && service.supervision.delay != 0 ) - supervisor.scheduleWithFixedDelay( si.serviceName, ( Runnable ) instance, + supervisor.scheduleWithFixedDelay( si, ( Runnable ) instance, service.supervision.delay, MILLISECONDS ); } } diff --git a/oap-application/oap-application/src/main/java/oap/application/ModuleItem.java b/oap-application/oap-application/src/main/java/oap/application/ModuleItem.java index 04e3c03f7f..f070d0e601 100644 --- a/oap-application/oap-application/src/main/java/oap/application/ModuleItem.java +++ b/oap-application/oap-application/src/main/java/oap/application/ModuleItem.java @@ -76,7 +76,7 @@ public boolean equals( Object o ) { if( this == o ) return true; if( o == null || getClass() != o.getClass() ) return false; - var that = ( ModuleItem ) o; + ModuleItem that = ( ModuleItem ) o; return module.name.equals( that.module.name ); } @@ -158,7 +158,7 @@ public boolean equals( Object o ) { if( this == o ) return true; if( o == null || getClass() != o.getClass() ) return false; - var that = ( ServiceItem ) o; + ServiceItem that = ( ServiceItem ) o; if( !moduleItem.module.name.equals( that.moduleItem.module.name ) ) return false; return serviceName.equals( that.serviceName ); @@ -172,7 +172,7 @@ public int hashCode() { } public void addDependsOn( ServiceReference serviceReference ) { - var found = Lists.find2( dependsOn, d -> d.equals( serviceReference ) ); + ServiceReference found = Lists.find2( dependsOn, d -> d.equals( serviceReference ) ); if( found == null || found.required ) { if( found != null ) dependsOn.remove( found ); dependsOn.add( serviceReference ); @@ -201,7 +201,7 @@ public boolean equals( Object o ) { if( this == o ) return true; if( o == null || getClass() != o.getClass() ) return false; - var that = ( ServiceReference ) o; + ServiceReference that = ( ServiceReference ) o; return serviceItem.serviceName.equals( that.serviceItem.serviceName ); } diff --git a/oap-application/oap-application/src/main/java/oap/application/supervision/Supervisor.java b/oap-application/oap-application/src/main/java/oap/application/supervision/Supervisor.java index 7ddff9e540..664ba1611a 100644 --- a/oap-application/oap-application/src/main/java/oap/application/supervision/Supervisor.java +++ b/oap-application/oap-application/src/main/java/oap/application/supervision/Supervisor.java @@ -26,6 +26,7 @@ import lombok.extern.slf4j.Slf4j; import oap.application.ApplicationConfiguration; import oap.application.KernelHelper; +import oap.application.ModuleItem; import oap.concurrent.Executors; import oap.util.BiStream; import oap.util.Dates; @@ -44,8 +45,8 @@ @Slf4j public class Supervisor { - private final LinkedHashMap supervised = new LinkedHashMap<>(); - private final LinkedHashMap> wrappers = new LinkedHashMap<>(); + private final LinkedHashMap supervised = new LinkedHashMap<>(); + private final LinkedHashMap> wrappers = new LinkedHashMap<>(); private boolean stopped = false; @@ -82,34 +83,34 @@ private static void runAndDetectTimeout( String name, ShutdownConfiguration shut } } - public synchronized void startSupervised( String name, Object service, + public synchronized void startSupervised( ModuleItem.ServiceItem si, Object service, List preStartWith, List startWith, List preStopWith, List stopWith ) { - this.supervised.put( name, new StartableService( service, preStartWith, startWith, preStopWith, stopWith ) ); + this.supervised.put( si, new StartableService( service, preStartWith, startWith, preStopWith, stopWith ) ); } // public synchronized void startScheduledThread( String name, Object instance, long delay, TimeUnit milliseconds ) { // this.wrappers.put( name, new ThreadService( name, ( Runnable ) instance, this ) ); // } - public synchronized void startThread( String name, Object instance, ApplicationConfiguration.ModuleShutdown shutdown ) { - this.wrappers.put( name, new ThreadService( name, ( Runnable ) instance, this, shutdown ) ); + public synchronized void startThread( ModuleItem.ServiceItem si, Object instance, ApplicationConfiguration.ModuleShutdown shutdown ) { + this.wrappers.put( si, new ThreadService( si.toString(), ( Runnable ) instance, this, shutdown ) ); } - public synchronized void scheduleWithFixedDelay( String name, Runnable service, long delay, TimeUnit unit ) { - this.wrappers.put( name, new DelayScheduledService( service, delay, unit ) ); + public synchronized void scheduleWithFixedDelay( ModuleItem.ServiceItem si, Runnable service, long delay, TimeUnit unit ) { + this.wrappers.put( si, new DelayScheduledService( service, delay, unit ) ); } - public synchronized void scheduleCron( String name, Runnable service, String cron ) { - this.wrappers.put( name, new CronScheduledService( service, cron ) ); + public synchronized void scheduleCron( ModuleItem.ServiceItem si, Runnable service, String cron ) { + this.wrappers.put( si, new CronScheduledService( service, cron ) ); } public synchronized void preStart() { log.debug( "pre starting..." ); - this.supervised.forEach( ( name, service ) -> { - log.debug( "pre starting {}...", name ); - KernelHelper.setThreadNameSuffix( name ); + this.supervised.forEach( ( si, service ) -> { + log.debug( "pre starting {}...", si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.preStart(); } finally { @@ -119,45 +120,45 @@ public synchronized void preStart() { BiStream.of( this.wrappers ) .reversed() - .forEach( ( name, service ) -> { - log.debug( "[{}] pre starting {}...", service.type(), name ); - KernelHelper.setThreadNameSuffix( name ); + .forEach( ( si, service ) -> { + log.debug( "[{}] pre starting {}...", service.type(), si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.preStart(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "[{}] pre starting {}... Done.", service.type(), name ); + log.debug( "[{}] pre starting {}... Done.", service.type(), si ); } ); } public synchronized void start() { log.debug( "starting..." ); this.stopped = false; - this.supervised.forEach( ( name, service ) -> { - log.debug( "starting {}...", name ); + this.supervised.forEach( ( si, service ) -> { + log.debug( "starting {}...", si ); long start = System.currentTimeMillis(); - KernelHelper.setThreadNameSuffix( name ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.start(); } finally { KernelHelper.restoreThreadName(); } long end = System.currentTimeMillis(); - log.debug( "starting {}... Done. ({}ms)", name, end - start ); + log.debug( "starting {}... Done. ({}ms)", si, end - start ); } ); - this.wrappers.forEach( ( name, service ) -> { - log.debug( "[{}] starting {}...", service.type(), name ); + this.wrappers.forEach( ( si, service ) -> { + log.debug( "[{}] starting {}...", service.type(), si ); long start = System.currentTimeMillis(); - KernelHelper.setThreadNameSuffix( name ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.start(); } finally { KernelHelper.restoreThreadName(); } long end = System.currentTimeMillis(); - log.debug( "[{}] starting {}... Done. ({}ms)", service.type(), name, end - start ); + log.debug( "[{}] starting {}... Done. ({}ms)", service.type(), si, end - start ); } ); } @@ -168,36 +169,36 @@ public synchronized void preStop( ApplicationConfiguration.ModuleShutdown shutdo BiStream.of( this.wrappers ) .reversed() - .forEach( ( name, service ) -> { + .forEach( ( si, service ) -> { Runnable func = () -> { - log.debug( "[{}] pre stopping {}...", service.type(), name ); - KernelHelper.setThreadNameSuffix( name ); + log.debug( "[{}] pre stopping {}...", service.type(), si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.preStop(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "[{}] pre stopping {}... Done.", service.type(), name ); + log.debug( "[{}] pre stopping {}... Done.", service.type(), si ); }; - runAndDetectTimeout( name, shutdownConfiguration, func ); + runAndDetectTimeout( si.toString(), shutdownConfiguration, func ); } ); BiStream.of( this.supervised ) .reversed() - .forEach( ( name, service ) -> { + .forEach( ( si, service ) -> { Runnable func = () -> { - log.debug( "pre stopping {}...", name ); - KernelHelper.setThreadNameSuffix( name ); + log.debug( "pre stopping {}...", si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.preStop(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "pre stopping {}... Done.", name ); + log.debug( "pre stopping {}... Done.", si ); }; - runAndDetectTimeout( name, shutdownConfiguration, func ); + runAndDetectTimeout( si.toString(), shutdownConfiguration, func ); } ); } } @@ -211,37 +212,37 @@ public synchronized void stop( ApplicationConfiguration.ModuleShutdown shutdown BiStream.of( this.wrappers ) .reversed() - .forEach( ( name, service ) -> { + .forEach( ( si, service ) -> { Runnable func = () -> { - log.debug( "[{}] stopping {}...", service.type(), name ); - KernelHelper.setThreadNameSuffix( name ); + log.debug( "[{}] stopping {}...", service.type(), si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.stop(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "[{}] stopping {}... Done.", service.type(), name ); + log.debug( "[{}] stopping {}... Done.", service.type(), si ); }; - runAndDetectTimeout( name, shutdownConfiguration, func ); + runAndDetectTimeout( si.toString(), shutdownConfiguration, func ); } ); this.wrappers.clear(); BiStream.of( this.supervised ) .reversed() - .forEach( ( name, service ) -> { + .forEach( ( si, service ) -> { Runnable func = () -> { - log.debug( "stopping {}...", name ); - KernelHelper.setThreadNameSuffix( name ); + log.debug( "stopping {}...", si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.stop(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "stopping {}... Done.", name ); + log.debug( "stopping {}... Done.", si ); }; - runAndDetectTimeout( name, shutdownConfiguration, func ); + runAndDetectTimeout( si.toString(), shutdownConfiguration, func ); } ); this.supervised.clear(); } @@ -255,40 +256,40 @@ public synchronized void stop( String serviceName, ApplicationConfiguration.Modu try( ShutdownConfiguration shutdownConfiguration = new ShutdownConfiguration( shutdown ) ) { BiStream.of( this.wrappers ) - .filter( ( name, _ ) -> name.equals( serviceName ) ) - .forEach( ( name, service ) -> { + .filter( ( si, _ ) -> si.serviceName.equals( serviceName ) ) + .forEach( ( si, service ) -> { Runnable func = () -> { - log.debug( "[{}] stopping {}...", service.type(), name ); - KernelHelper.setThreadNameSuffix( name ); + log.debug( "[{}] stopping {}...", service.type(), si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.preStop(); service.stop(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "[{}] stopping {}... Done.", service.type(), name ); + log.debug( "[{}] stopping {}... Done.", service.type(), si ); }; - runAndDetectTimeout( name, shutdownConfiguration, func ); + runAndDetectTimeout( si.toString(), shutdownConfiguration, func ); } ); this.wrappers.clear(); BiStream.of( this.supervised ) - .filter( ( name, _ ) -> name.equals( serviceName ) ) - .forEach( ( name, service ) -> { + .filter( ( si, _ ) -> si.serviceName.equals( serviceName ) ) + .forEach( ( si, service ) -> { Runnable func = () -> { - log.debug( "stopping {}...", name ); - KernelHelper.setThreadNameSuffix( name ); + log.debug( "stopping {}...", si ); + KernelHelper.setThreadNameSuffix( si.toString() ); try { service.preStop(); service.stop(); } finally { KernelHelper.restoreThreadName(); } - log.debug( "stopping {}... Done.", name ); + log.debug( "stopping {}... Done.", si ); }; - runAndDetectTimeout( name, shutdownConfiguration, func ); + runAndDetectTimeout( si.toString(), shutdownConfiguration, func ); } ); } } diff --git a/oap-formats/oap-logstream/oap-logstream-test/src/test/java/oap/logstream/disk/RowBinaryWriterTest.java b/oap-formats/oap-logstream/oap-logstream-test/src/test/java/oap/logstream/disk/RowBinaryWriterTest.java index 5221ce919d..4147a04fcc 100644 --- a/oap-formats/oap-logstream/oap-logstream-test/src/test/java/oap/logstream/disk/RowBinaryWriterTest.java +++ b/oap-formats/oap-logstream/oap-logstream-test/src/test/java/oap/logstream/disk/RowBinaryWriterTest.java @@ -88,7 +88,7 @@ public void testWrite() throws IOException { LogId logId = new LogId( "", "log", "log", Map.of( "p", "1" ), headers, types ); Path logs = testDirectoryFixture.testPath( "logs" ); - try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost" ) ) { + try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost", null ) ) { writer.write( CURRENT_PROTOCOL_VERSION, content1 ); writer.write( CURRENT_PROTOCOL_VERSION, content2 ); } @@ -124,7 +124,7 @@ public void testConcurrency() throws IOException { int count = 10; - try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost" ) ) { + try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost", null ) ) { try( ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor() ) { for( long i = 0; i < count; i++ ) { @@ -168,7 +168,7 @@ public void testWriteToNewVersionWhenCompleted() throws IOException { Path v1 = logs.resolve( "1-file-02-47b82ddc0-1.rb.gz.rb.gz" ); Path v2 = logs.resolve( "1-file-02-47b82ddc0-2.rb.gz.rb.gz" ); - try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost" ) ) { + try( RowBinaryWriter writer = new RowBinaryWriter( templateEngineFixture.templateEngine, logs, FILE_PATTERN, logId, 1024, BPH_12, 20, "localhost", null ) ) { writer.write( CURRENT_PROTOCOL_VERSION, content1 ); writer.refresh(); diff --git a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/AbstractWriter.java b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/AbstractWriter.java index 65abfd06cd..d685a121fb 100644 --- a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/AbstractWriter.java +++ b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/AbstractWriter.java @@ -28,7 +28,6 @@ import io.micrometer.core.instrument.Metrics; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; -import oap.concurrent.Stopwatch; import oap.logstream.LogId; import oap.logstream.LogIdTemplate; import oap.logstream.LogStreamProtocol.ProtocolVersion; @@ -52,9 +51,9 @@ public abstract class AbstractWriter implements Closeable { protected final LogId logId; protected final Timestamp timestamp; protected final int bufferSize; - protected final Stopwatch stopwatch = new Stopwatch(); protected final int maxVersions; protected final String hostname; + protected final FileWriterNotification notification; protected final ReentrantLock lock = new ReentrantLock(); protected LogFile logFile; protected String lastPattern; @@ -62,13 +61,14 @@ public abstract class AbstractWriter implements Closeable { protected boolean closed = false; protected AbstractWriter( TemplateEngine templateEngine, LogFormat logFormat, Path logDirectory, String filePattern, LogId logId, int bufferSize, Timestamp timestamp, - int maxVersions, String hostname ) { + int maxVersions, String hostname, FileWriterNotification notification ) { this.templateEngine = templateEngine; this.logFormat = logFormat; this.logDirectory = logDirectory; this.filePattern = filePattern; this.maxVersions = maxVersions; this.hostname = hostname; + this.notification = notification; log.trace( "filePattern {}", filePattern ); Preconditions.checkArgument( filePattern.matches( ".*[${]\\{\\s*LOG_VERSION\\s*}}?.*" ), "file pattern must contains LOG_VERSION variable" ); @@ -120,12 +120,14 @@ public void refresh() { public void refresh( boolean forceSync ) { lock.lock(); try { - log.debug( "refresh {}...", lastPattern ); + if( logFile != null ) { + log.debug( "refresh {}...", lastPattern ); + } String currentPattern = currentPattern(); if( forceSync || !Objects.equals( this.lastPattern, currentPattern ) ) { - log.debug( "lastPattern {} currentPattern {} version {}", lastPattern, currentPattern, fileVersion ); + log.trace( "lastPattern {} currentPattern {} version {}", lastPattern, currentPattern, fileVersion ); String patternWithPreviousVersion = currentPattern( fileVersion - 1 ); if( !Objects.equals( patternWithPreviousVersion, this.lastPattern ) ) { @@ -138,7 +140,7 @@ public void refresh( boolean forceSync ) { lastPattern = currentPattern; } else { - log.debug( "refresh {}... SKIP", lastPattern ); + log.trace( "refresh {}... SKIP", lastPattern ); } } finally { lock.unlock(); @@ -153,14 +155,17 @@ protected void closeOutput() throws LoggerException { lock.lock(); try { if( logFile != null ) try { - stopwatch.count( logFile::close ); + logFile.close(); long fileSize = logFile.getDataSize(); log.trace( "closing output {} ({} bytes)", this, fileSize ); Metrics.summary( "logstream_logging_server_bucket_size" ).record( fileSize ); - Metrics.summary( "logstream_logging_server_bucket_time_seconds" ).record( Dates.nanosToSeconds( stopwatch.elapsed() ) ); logFile.readyForUpload(); + + if( notification != null ) { + notification.fileClosed( logFile.outFilename ); + } } finally { logFile = null; } @@ -173,7 +178,9 @@ protected void closeOutput() throws LoggerException { public void close() { lock.lock(); try { - log.debug( "closing {}", this ); + if( logFile != null ) { + log.debug( "closing {}", this ); + } closed = true; closeOutput(); } finally { diff --git a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/DiskLoggerBackend.java b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/DiskLoggerBackend.java index ce50188250..d46804b502 100644 --- a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/DiskLoggerBackend.java +++ b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/DiskLoggerBackend.java @@ -63,6 +63,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import static java.util.concurrent.TimeUnit.MILLISECONDS; @@ -75,7 +76,7 @@ *
  • POD_NAME
    • */ @Slf4j -public class DiskLoggerBackend extends AbstractLoggerBackend implements Cloneable, AutoCloseable { +public class DiskLoggerBackend extends AbstractLoggerBackend implements FileWriterNotification, Cloneable, AutoCloseable { public static final int DEFAULT_BUFFER = 1024 * 100; public static final long DEFAULT_FREE_SPACE_REQUIRED = 2000000000L; public final LinkedHashMap filePatternByType = new LinkedHashMap<>(); @@ -93,6 +94,7 @@ public class DiskLoggerBackend extends AbstractLoggerBackend implements Cloneabl public long refreshInitDelay = Dates.s( 10 ); public long refreshPeriod = Dates.s( 10 ); public volatile boolean closed; + protected CopyOnWriteArrayList notifications = new CopyOnWriteArrayList<>(); public DiskLoggerBackend( TemplateEngine templateEngine, Path logDirectory, Timestamp timestamp, int bufferSize, String hostname ) { this( templateEngine, logDirectory, new WriterConfiguration(), timestamp, bufferSize, hostname ); @@ -116,8 +118,8 @@ public DiskLoggerBackend( TemplateEngine templateEngine, Path logDirectory, Writ this.writers = CacheBuilder.newBuilder() .ticker( JodaTicker.JODA_TICKER ) .expireAfterAccess( 60 / timestamp.bucketsPerHour * 3, TimeUnit.MINUTES ) - .removalListener( notification -> { - Closeables.close( ( Closeable ) notification.getValue() ); + .removalListener( l -> { + Closeables.close( ( Closeable ) l.getValue() ); } ) .build( new CacheLoader<>() { @Override @@ -126,7 +128,7 @@ public AbstractWriter load( LogId id ) { log.trace( "new writer id '{}' filePattern '{}'", id, fp ); - return new RowBinaryWriter( templateEngine, DiskLoggerBackend.this.logDirectory, fp.path, id, bufferSize, timestamp, maxVersions, hostname ); + return new RowBinaryWriter( templateEngine, DiskLoggerBackend.this.logDirectory, fp.path, id, bufferSize, timestamp, maxVersions, hostname, DiskLoggerBackend.this ); } } ); Metrics.gauge( "logstream_logging_disk_writers", List.of( Tag.of( "path", this.logDirectory.toString() ) ), @@ -243,6 +245,17 @@ public String toString() { .toString(); } + @Override + public void fileClosed( Path outFilename ) { + for( FileWriterNotification notification : notifications ) { + notification.fileClosed( outFilename ); + } + } + + public void addNotificationListener( FileWriterNotification notification ) { + this.notifications.add( notification ); + } + @ToString @EqualsAndHashCode public static class FilePatternConfiguration { diff --git a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/FileWriterNotification.java b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/FileWriterNotification.java new file mode 100644 index 0000000000..eb245f114c --- /dev/null +++ b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/FileWriterNotification.java @@ -0,0 +1,7 @@ +package oap.logstream.disk; + +import java.nio.file.Path; + +public interface FileWriterNotification { + void fileClosed( Path outFilename ); +} diff --git a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/RowBinaryWriter.java b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/RowBinaryWriter.java index 9aeda25d3c..0bf8336964 100644 --- a/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/RowBinaryWriter.java +++ b/oap-formats/oap-logstream/oap-logstream/src/main/java/oap/logstream/disk/RowBinaryWriter.java @@ -18,8 +18,9 @@ @Slf4j public class RowBinaryWriter extends AbstractWriter { - public RowBinaryWriter( TemplateEngine templateEngine, Path logDirectory, String filePattern, LogId logId, int bufferSize, Timestamp timestamp, int maxVersions, String hostname ) { - super( templateEngine, LogFormat.ROW_BINARY_GZ, logDirectory, filePattern, logId, bufferSize, timestamp, maxVersions, hostname ); + public RowBinaryWriter( TemplateEngine templateEngine, Path logDirectory, String filePattern, LogId logId, + int bufferSize, Timestamp timestamp, int maxVersions, String hostname, FileWriterNotification notification ) { + super( templateEngine, LogFormat.ROW_BINARY_GZ, logDirectory, filePattern, logId, bufferSize, timestamp, maxVersions, hostname, notification ); } @Override diff --git a/oap-notification/oap-notification-client/pom.xml b/oap-notification/oap-notification-client/pom.xml new file mode 100644 index 0000000000..2bc20ca37f --- /dev/null +++ b/oap-notification/oap-notification-client/pom.xml @@ -0,0 +1,27 @@ + + + 4.0.0 + + + oap + oap-notification + ${oap.project.version} + + + oap-notification-client + + + + oap + oap-stdlib + ${project.version} + + + + org.projectlombok + lombok + + + diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/MockNotificationTransport.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/MockNotificationTransport.java new file mode 100644 index 0000000000..edf0c87eb1 --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/MockNotificationTransport.java @@ -0,0 +1,20 @@ +package oap.notification; + +import lombok.extern.slf4j.Slf4j; +import oap.json.Binder; + +import java.util.List; +import java.util.function.Consumer; + +@Slf4j +public class MockNotificationTransport implements NotificationTransport { + @Override + public void publish( String topic, Qos qos, Notification notification ) { + log.trace( "publish topic {} qos {} notification {}", topic, qos, Binder.json.marshal( notification ) ); + } + + @Override + public void subscribe( List topics, Consumer notificationConsumer ) { + log.trace( "subscribe topics {}", topics ); + } +} diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/Notification.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/Notification.java new file mode 100644 index 0000000000..31aeb0046a --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/Notification.java @@ -0,0 +1,27 @@ +package oap.notification; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import com.fasterxml.jackson.databind.annotation.JsonTypeIdResolver; +import oap.json.TypeIdFactory; + +import java.io.Serial; +import java.io.Serializable; + +public class Notification implements Serializable { + @Serial + private static final long serialVersionUID = -1730908173571715179L; + + @JsonTypeIdResolver( TypeIdFactory.class ) + @JsonTypeInfo( use = JsonTypeInfo.Id.CUSTOM, include = JsonTypeInfo.As.EXTERNAL_PROPERTY, property = "object:type" ) + public final Serializable message; + + @JsonCreator + public Notification( Serializable message ) { + this.message = message; + } + + public Notification( Notification notification ) { + this( notification.message ); + } +} diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationException.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationException.java new file mode 100644 index 0000000000..84e3da9a95 --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationException.java @@ -0,0 +1,18 @@ +package oap.notification; + +public class NotificationException extends RuntimeException { + public NotificationException() { + } + + public NotificationException( String message ) { + super( message ); + } + + public NotificationException( String message, Throwable cause ) { + super( message, cause ); + } + + public NotificationException( Throwable cause ) { + super( cause ); + } +} diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationPublish.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationPublish.java new file mode 100644 index 0000000000..47a14212cc --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationPublish.java @@ -0,0 +1,26 @@ +package oap.notification; + +import lombok.ToString; + +import java.io.Serial; +import java.io.Serializable; + +@ToString +public class NotificationPublish extends Notification { + @Serial + private static final long serialVersionUID = 8509736862218143643L; + + public final String topic; + + public NotificationPublish( String topic, Notification notification ) { + super( notification ); + + this.topic = topic; + } + + public NotificationPublish( String topic, Serializable message ) { + super( message ); + + this.topic = topic; + } +} diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationService.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationService.java new file mode 100644 index 0000000000..16d1eff012 --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationService.java @@ -0,0 +1,25 @@ +package oap.notification; + +import java.io.Serializable; +import java.util.List; +import java.util.function.Consumer; + +public class NotificationService { + public final NotificationTransport notificationTransport; + + public NotificationService( NotificationTransport notificationTransport ) { + this.notificationTransport = notificationTransport; + } + + public void sendNotification( String topic, Qos qos, TMessage message ) throws NotificationException { + notificationTransport.publish( topic, qos, new Notification( message ) ); + } + + public void subscribeToTopic( String topic, Consumer notificationConsumer ) { + notificationTransport.subscribe( topic, notificationConsumer ); + } + + public void subscribeToTopic( List topics, Consumer notificationConsumer ) { + notificationTransport.subscribe( topics, notificationConsumer ); + } +} diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationTransport.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationTransport.java new file mode 100644 index 0000000000..184714001b --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/NotificationTransport.java @@ -0,0 +1,14 @@ +package oap.notification; + +import java.util.List; +import java.util.function.Consumer; + +public interface NotificationTransport { + void publish( String topic, Qos qos, Notification notification ) throws NotificationException; + + default void subscribe( String topic, Consumer notificationConsumer ) { + subscribe( List.of( topic ), notificationConsumer ); + } + + void subscribe( List topics, Consumer notificationConsumer ); +} diff --git a/oap-notification/oap-notification-client/src/main/java/oap/notification/Qos.java b/oap-notification/oap-notification-client/src/main/java/oap/notification/Qos.java new file mode 100644 index 0000000000..e79d5fa1bc --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/java/oap/notification/Qos.java @@ -0,0 +1,7 @@ +package oap.notification; + +public enum Qos { + AT_MOST_ONCE, + AT_LEAST_ONCE, + EXACTLY_ONCE +} diff --git a/oap-notification/oap-notification-client/src/main/resources/META-INF/oap-module.oap b/oap-notification/oap-notification-client/src/main/resources/META-INF/oap-module.oap new file mode 100644 index 0000000000..534190fdd9 --- /dev/null +++ b/oap-notification/oap-notification-client/src/main/resources/META-INF/oap-module.oap @@ -0,0 +1,20 @@ +name = oap-notification + +services { + mock-notification-transport { + implementation = oap.notification.MockNotificationTransport + } + + notification-transport { + abstract = true + implementation = oap.notification.NotificationTransport + default = + } + + notification { + implementation = oap.notification.NotificationService + parameters { + notificationTransport = + } + } +} diff --git a/oap-notification/oap-notification-mqtt/pom.xml b/oap-notification/oap-notification-mqtt/pom.xml new file mode 100644 index 0000000000..31417ba9aa --- /dev/null +++ b/oap-notification/oap-notification-mqtt/pom.xml @@ -0,0 +1,38 @@ + + + 4.0.0 + + + oap + oap-notification + ${oap.project.version} + + + oap-notification-mqtt + + + + com.hivemq + hivemq-mqtt-client + ${oap.deps.hivemq-mqtt-client.version} + + + + oap + oap-notification-client + ${parent.version} + + + oap + oap-application-annotation + ${parent.version} + + + + org.projectlombok + lombok + + + diff --git a/oap-notification/oap-notification-mqtt/src/main/java/oap/notification/mqtt/HivemqNotificationTransport.java b/oap-notification/oap-notification-mqtt/src/main/java/oap/notification/mqtt/HivemqNotificationTransport.java new file mode 100644 index 0000000000..d0d4038b7d --- /dev/null +++ b/oap-notification/oap-notification-mqtt/src/main/java/oap/notification/mqtt/HivemqNotificationTransport.java @@ -0,0 +1,127 @@ +package oap.notification.mqtt; + +import com.hivemq.client.mqtt.MqttClient; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishResult; +import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5Subscription; +import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck; +import lombok.extern.slf4j.Slf4j; +import oap.application.annotation.Start; +import oap.application.annotation.Stop; +import oap.json.Binder; +import oap.notification.Notification; +import oap.notification.NotificationException; +import oap.notification.NotificationPublish; +import oap.notification.NotificationTransport; +import oap.notification.Qos; +import oap.util.Dates; +import oap.util.Lists; + +import java.util.List; +import java.util.concurrent.CompletionException; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +@Slf4j +public class HivemqNotificationTransport implements NotificationTransport, AutoCloseable { + private final String identifier; + private final String host; + private final int port; + public long connectTimeout = Dates.s( 10 ); + public long publishTimeout = Dates.s( 1 ); + private Mqtt5AsyncClient client; + + public HivemqNotificationTransport( String identifier, String host, int port ) { + this.identifier = identifier; + this.host = host; + this.port = port; + } + + @Start + public void start() { + client = MqttClient + .builder() + .useMqttVersion5() + .identifier( identifier ) + .serverHost( host ) + .serverPort( port ) + + .automaticReconnect() + .applyAutomaticReconnect() + + .buildAsync(); + + Mqtt5ConnAck ack = client + .connectWith() + .send() + .orTimeout( connectTimeout, TimeUnit.MILLISECONDS ) + .join(); + + log.debug( "Connected to MQTT server at {}:{} response {}", host, port, ack ); + } + + @Stop + public void close() { + if( client != null && client.getState().isConnectedOrReconnect() ) { + try { + client + .disconnectWith() + .send() + .orTimeout( connectTimeout, TimeUnit.MILLISECONDS ) + .join(); + } catch( CompletionException e ) { + log.error( e.getCause().getMessage() ); + } + } + } + + @Override + public void publish( String topic, Qos qos, Notification notification ) throws NotificationException { + try { + log.trace( "publish topic {} qos {} notification {}", topic, qos, Binder.json.marshal( notification ) ); + + Mqtt5PublishResult result = client + .publishWith() + .topic( topic ) + .qos( convertQos( qos ) ) + .payload( Binder.json.marshal( notification ).getBytes() ) + .send() + .orTimeout( publishTimeout, TimeUnit.MILLISECONDS ) + .join(); + + log.trace( "publish topic {} qos {} result {}", topic, qos, result ); + } catch( CompletionException e ) { + throw new NotificationException( e.getCause() ); + } + } + + @Override + public void subscribe( List topics, Consumer notificationConsumer ) { + Mqtt5SubAck ack = client + .subscribeWith() + .addSubscriptions( Lists.map( topics, topic -> Mqtt5Subscription.builder().topicFilter( topic ).build() ) ) + .callback( mqtt5Publish -> { + byte[] payloadAsBytes = mqtt5Publish.getPayloadAsBytes(); + + log.trace( "topic {} payload {}", mqtt5Publish.getTopic(), + payloadAsBytes.length > 0 ? new String( payloadAsBytes ) : "" ); + + notificationConsumer.accept( new NotificationPublish( mqtt5Publish.getTopic().toString(), Binder.json.unmarshal( Notification.class, payloadAsBytes ) ) ); + } ) + .send() + .orTimeout( publishTimeout, TimeUnit.MILLISECONDS ) + .join(); + + log.trace( "publish topics {} result {}", topics, ack ); + } + + private MqttQos convertQos( Qos qos ) { + return switch( qos ) { + case AT_MOST_ONCE -> MqttQos.AT_MOST_ONCE; + case EXACTLY_ONCE -> MqttQos.EXACTLY_ONCE; + case AT_LEAST_ONCE -> MqttQos.AT_LEAST_ONCE; + }; + } +} diff --git a/oap-notification/oap-notification-mqtt/src/main/resources/META-INF/oap-module.oap b/oap-notification/oap-notification-mqtt/src/main/resources/META-INF/oap-module.oap new file mode 100644 index 0000000000..d86aba6dbc --- /dev/null +++ b/oap-notification/oap-notification-mqtt/src/main/resources/META-INF/oap-module.oap @@ -0,0 +1,15 @@ +name = oap-notification-mqtt + +dependsOn = oap-notification + +services { + mqtt-notification-transport { + implementation = oap.notification.mqtt.HivemqNotificationTransport + parameters { +// identifier = OAP + host = "" + port = 1883 + } + supervision.supervise = true + } +} diff --git a/oap-notification/oap-notification-test/pom.xml b/oap-notification/oap-notification-test/pom.xml new file mode 100644 index 0000000000..aa05b9d742 --- /dev/null +++ b/oap-notification/oap-notification-test/pom.xml @@ -0,0 +1,32 @@ + + + 4.0.0 + + + oap + oap-notification + ${oap.project.version} + + + oap-notification-test + + + + oap + oap-notification-mqtt + ${project.version} + + + oap + oap-stdlib-test + ${project.version} + + + + org.projectlombok + lombok + + + diff --git a/oap-notification/oap-notification-test/src/main/java/oap/notification/mqtt/MosquittoFixture.java b/oap-notification/oap-notification-test/src/main/java/oap/notification/mqtt/MosquittoFixture.java new file mode 100644 index 0000000000..5d6d867d06 --- /dev/null +++ b/oap-notification/oap-notification-test/src/main/java/oap/notification/mqtt/MosquittoFixture.java @@ -0,0 +1,47 @@ +package oap.notification.mqtt; + +import com.github.dockerjava.api.model.ExposedPort; +import com.github.dockerjava.api.model.PortBinding; +import com.github.dockerjava.api.model.Ports; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import oap.testng.AbstractFixture; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.utility.DockerImageName; + +@Slf4j +public class MosquittoFixture extends AbstractFixture { + private static final String VERSION = "2.1.2-alpine"; + @Getter + private final int port; + private GenericContainer container; + + public MosquittoFixture() { + port = definePort( "MQTT_PORT" ); + } + + @Override + protected void before() { + super.before(); + + PortBinding portBinding = new PortBinding( + Ports.Binding.bindPort( port ), + new ExposedPort( 1883 ) ); + + container = new GenericContainer<>( DockerImageName.parse( "eclipse-mosquitto:" + VERSION ) ) + .withExposedPorts( 1883 ) + .withCreateContainerCmdModifier( cmd -> cmd.getHostConfig().withPortBindings( portBinding ) ) + .withLogConsumer( new Slf4jLogConsumer( log ) ); + container.start(); + } + + @Override + protected void after() { + if( container != null ) { + container.stop(); + } + + super.after(); + } +} diff --git a/oap-notification/oap-notification-test/src/test/java/oap/notification/TestNotificationMessage.java b/oap-notification/oap-notification-test/src/test/java/oap/notification/TestNotificationMessage.java new file mode 100644 index 0000000000..d23f4e5444 --- /dev/null +++ b/oap-notification/oap-notification-test/src/test/java/oap/notification/TestNotificationMessage.java @@ -0,0 +1,18 @@ +package oap.notification; + +import lombok.AllArgsConstructor; +import lombok.EqualsAndHashCode; +import lombok.ToString; + +import java.io.Serial; +import java.io.Serializable; + +@EqualsAndHashCode +@ToString +@AllArgsConstructor +public class TestNotificationMessage implements Serializable { + @Serial + private static final long serialVersionUID = -4166315788080834194L; + + public final String value; +} diff --git a/oap-notification/oap-notification-test/src/test/java/oap/notification/mqtt/MosquittoNotificationServiceTest.java b/oap-notification/oap-notification-test/src/test/java/oap/notification/mqtt/MosquittoNotificationServiceTest.java new file mode 100644 index 0000000000..ff611ace6f --- /dev/null +++ b/oap-notification/oap-notification-test/src/test/java/oap/notification/mqtt/MosquittoNotificationServiceTest.java @@ -0,0 +1,45 @@ +package oap.notification.mqtt; + +import oap.notification.NotificationService; +import oap.notification.Qos; +import oap.notification.TestNotificationMessage; +import oap.testng.Fixtures; +import org.testng.annotations.Test; + +import java.util.StringJoiner; + +import static org.assertj.core.api.Assertions.assertThat; + +public class MosquittoNotificationServiceTest extends Fixtures { + private final MosquittoFixture mosquittoFixture; + + public MosquittoNotificationServiceTest() { + mosquittoFixture = fixture( new MosquittoFixture() ); + } + + @Test + public void testMessages() { + StringJoiner msg = new StringJoiner( " / " ); + + try( HivemqNotificationTransport notificationTransportClient1 = new HivemqNotificationTransport( "client1", "127.0.0.1", mosquittoFixture.getPort() ); + HivemqNotificationTransport notificationTransportClient2 = new HivemqNotificationTransport( "client2", "127.0.0.1", mosquittoFixture.getPort() ) ) { + + notificationTransportClient1.start(); + notificationTransportClient2.start(); + + NotificationService notificationService1 = new NotificationService( notificationTransportClient1 ); + NotificationService notificationService2 = new NotificationService( notificationTransportClient2 ); + + notificationService1.sendNotification( "/test", Qos.AT_LEAST_ONCE, new TestNotificationMessage( "val1" ) ); + + notificationService2.subscribeToTopic( "/test", notification -> { + TestNotificationMessage notificationMessage = ( TestNotificationMessage ) notification.message; + msg.add( notificationMessage.value ); + } ); + + notificationService1.sendNotification( "/test", Qos.AT_LEAST_ONCE, new TestNotificationMessage( "val2" ) ); + + assertThat( msg ).hasToString( "val2" ); + } + } +} diff --git a/oap-notification/oap-notification-test/src/test/resources/META-INF/oap-module.oap b/oap-notification/oap-notification-test/src/test/resources/META-INF/oap-module.oap new file mode 100644 index 0000000000..5a5e1f0526 --- /dev/null +++ b/oap-notification/oap-notification-test/src/test/resources/META-INF/oap-module.oap @@ -0,0 +1,14 @@ +name = oap-notification-test + +services { + +} + +configurations = [ + { + loader = oap.json.TypeIdFactory + config { + oap-test-notification = oap.notification.TestNotificationMessage + } + } +] \ No newline at end of file diff --git a/oap-notification/pom.xml b/oap-notification/pom.xml new file mode 100644 index 0000000000..a7f46d032d --- /dev/null +++ b/oap-notification/pom.xml @@ -0,0 +1,22 @@ + + + 4.0.0 + + pom + + + oap + oap + ${oap.project.version} + + + oap-notification + + + oap-notification-client + oap-notification-mqtt + oap-notification-test + + diff --git a/oap-stdlib-test/src/main/java/oap/testng/Asserts.java b/oap-stdlib-test/src/main/java/oap/testng/Asserts.java index ac05aa156e..81146d6044 100644 --- a/oap-stdlib-test/src/main/java/oap/testng/Asserts.java +++ b/oap-stdlib-test/src/main/java/oap/testng/Asserts.java @@ -42,6 +42,7 @@ import org.testng.Assert; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.net.URL; @@ -58,7 +59,7 @@ public final class Asserts { @SneakyThrows - public static void eventually( long retryTimeout, int retries, Try.ThrowingRunnable asserts ) { + public static void eventually( long retryTimeout, int retries, Try.ThrowingRunnable asserts, @Nullable Runnable onFailure ) { boolean passed = false; Throwable exception = null; int count = retries; @@ -71,6 +72,10 @@ public static void eventually( long retryTimeout, int retries, Try.ThrowingRunna exception = e; Threads.sleepSafely( retryTimeout ); count--; + + if( onFailure != null ) { + onFailure.run(); + } } } if( !passed ) @@ -79,12 +84,20 @@ public static void eventually( long retryTimeout, int retries, Try.ThrowingRunna } public static void assertEventually( long retryTimeout, int retries, oap.util.function.Try.ThrowingRunnable asserts ) { - eventually( retryTimeout, retries, asserts ); + assertEventually( retryTimeout, retries, asserts, null ); + } + + public static void assertEventually( long retryTimeout, int retries, oap.util.function.Try.ThrowingRunnable asserts, @Nullable Runnable onFailure ) { + eventually( retryTimeout, retries, asserts, onFailure ); } public static void assertEventually( Duration duration, Duration retryInterval, oap.util.function.Try.ThrowingRunnable asserts ) { + assertEventually( duration, retryInterval, asserts, null ); + } + + public static void assertEventually( Duration duration, Duration retryInterval, oap.util.function.Try.ThrowingRunnable asserts, @Nullable Runnable onFailure ) { int retries = ( int ) ( duration.toMillis() / retryInterval.toMillis() ); - eventually( retryInterval.toMillis(), retries, asserts ); + eventually( retryInterval.toMillis(), retries, asserts, onFailure ); } @Deprecated diff --git a/oap-stdlib/src/main/java/oap/json/Binder.java b/oap-stdlib/src/main/java/oap/json/Binder.java index 4274099177..9b6c79704c 100644 --- a/oap-stdlib/src/main/java/oap/json/Binder.java +++ b/oap-stdlib/src/main/java/oap/json/Binder.java @@ -89,6 +89,7 @@ import java.util.Optional; import java.util.Set; +import static java.nio.charset.StandardCharsets.UTF_8; import static oap.io.IoStreams.DEFAULT_BUFFER; import static oap.io.IoStreams.Encoding.from; @@ -277,6 +278,11 @@ private static String getLimitation( String json ) { return json; } + private static String getLimitation( byte[] json ) { + if( json != null && json.length > 20 ) return new String( json, UTF_8 ).substring( 0, 20 ) + "..."; + return json != null ? new String( json, UTF_8 ) : null; + } + public ObjectMapper getMapper() { return mapper; } @@ -560,6 +566,15 @@ public T unmarshal( Class clazz, String json ) throws JsonException { } } + public T unmarshal( Class clazz, byte[] json ) throws JsonException { + try { + return mapper.readValue( json, clazz ); + } catch( Exception e ) { + log.trace( "Cannot deserialize [{}] into {}", json, clazz.getCanonicalName() ); + throw new JsonException( "Cannot deserialize [" + getLimitation( json ) + "] to class: " + clazz.getCanonicalName(), e ); + } + } + public T unmarshal( Class clazz, Map map ) throws JsonException { try { return mapper.convertValue( map, clazz ); diff --git a/pom.xml b/pom.xml index 8a4c07f637..49fbf74d6f 100644 --- a/pom.xml +++ b/pom.xml @@ -33,6 +33,7 @@ oap-mcp oap-jpath oap-message + oap-notification oap-application oap-formats oap-storage @@ -66,7 +67,7 @@ - 25.9.2 + 25.9.3 25.0.1 25.0.0 @@ -123,6 +124,8 @@ 3.6.4 12.1.6 + 1.3.14 + 4.9.8