diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java index d83049abf..2c5411449 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.util.List; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.CoreModuleConfig; @@ -35,8 +36,6 @@ import org.apache.skywalking.oap.server.core.storage.model.IModelGetter; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.CollectionUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * TTL = Time To Live @@ -46,11 +45,10 @@ import org.slf4j.LoggerFactory; * override TTL, which could be more suitable for the implementation. No matter which TTL configurations are set, they * are all driven by this timer. */ +@Slf4j public enum DataTTLKeeperTimer { INSTANCE; - private static final Logger logger = LoggerFactory.getLogger(DataTTLKeeperTimer.class); - private ModuleManager moduleManager; private ClusterNodesQuery clusterNodesQuery; @@ -62,7 +60,7 @@ public enum DataTTLKeeperTimer { .scheduleAtFixedRate( new RunnableWithExceptionProtection( this::delete, - t -> logger.error("Remove data in background failure.", t) + t -> log.error("Remove data in background failure.", t) ), moduleConfig .getDataKeeperExecutePeriod(), moduleConfig.getDataKeeperExecutePeriod(), TimeUnit.MINUTES); } @@ -74,11 +72,11 @@ public enum DataTTLKeeperTimer { private void delete() { List remoteInstances = clusterNodesQuery.queryRemoteNodes(); if (CollectionUtils.isNotEmpty(remoteInstances) && !remoteInstances.get(0).getAddress().isSelf()) { - logger.info("The selected first getAddress is {}. Skip.", remoteInstances.get(0).toString()); + log.info("The selected first getAddress is {}. Skip.", remoteInstances.get(0).toString()); return; } - logger.info("Beginning to remove expired metrics from the storage."); + log.info("Beginning to remove expired metrics from the storage."); IModelGetter modelGetter = moduleManager.find(CoreModule.NAME).provider().getService(IModelGetter.class); List models = modelGetter.getModels(); models.forEach(model -> { @@ -95,8 +93,8 @@ public enum DataTTLKeeperTimer { .getService(IHistoryDeleteDAO.class) .deleteHistory(model, Metrics.TIME_BUCKET); } catch (IOException e) { - logger.warn("History of {} delete failure", model.getName()); - logger.error(e.getMessage(), e); + log.warn("History of {} delete failure", model.getName()); + log.error(e.getMessage(), e); } } } \ No newline at end of file diff --git a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/MultipleFilesChangeMonitor.java b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/MultipleFilesChangeMonitor.java new file mode 100644 index 000000000..8b4e2c47b --- /dev/null +++ b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/MultipleFilesChangeMonitor.java @@ -0,0 +1,242 @@ +/* + * 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.skywalking.oap.server.library.util; + +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * MultipleFilesChangeMonitor provides the capability to detect file or multiple files changed. It provide second level + * change detection and feedback mechanism. + * + * Due to memory cost, this monitor mechanism is not suitable for small files and usually being changed on the runtime + * by user manually or 3rd party tool. Typical, these files are config information or authentication files. + */ +@Slf4j +public class MultipleFilesChangeMonitor { + /** + * The backend scheduler to trigger all file monitoring. + */ + private static ScheduledFuture FILE_MONITOR_TASK_SCHEDULER; + private static ReentrantLock SCHEDULER_CHANGE_LOCK = new ReentrantLock(); + /** + * The list contains all monitors. + */ + private static List MONITOR_INSTANCES = new ArrayList<>(); + + /** + * The timestamp when last time do status checked. + */ + private long lastCheckTimestamp = 0; + /** + * The period of watching thread checking the file status. Unit is the second. + */ + private final long watchingPeriodInSec; + private List watchedFiles; + private FilesChangedNotifier notifier; + + /** + * Create a new monitor for the given files + * + * @param watchingPeriodInSec The check period. + * @param notifier to accept the file changed notification. + * @param files to be monitored. + */ + public MultipleFilesChangeMonitor(long watchingPeriodInSec, + FilesChangedNotifier notifier, + String... files) { + watchedFiles = new ArrayList<>(); + this.watchingPeriodInSec = watchingPeriodInSec; + this.notifier = notifier; + for (final String file : files) { + WatchedFile monitor = new WatchedFile(file); + watchedFiles.add(monitor); + } + } + + /** + * Check file changed status, if so, send the notification. + */ + private void checkAndNotify() { + if (System.currentTimeMillis() - lastCheckTimestamp < watchingPeriodInSec * 1000) { + // Don't reach the period threshold, ignore this check. + return; + } + + boolean isChanged = false; + for (final WatchedFile watchedFile : watchedFiles) { + isChanged = isChanged || watchedFile.detectContentChanged(); + } + if (isChanged) { + List contents = new ArrayList<>(watchedFiles.size()); + watchedFiles.forEach(file -> { + contents.add(file.fileContent); + }); + notifier.filesChanged(contents); + } + } + + /** + * One file changed will cause all related files loaded from the disk again with lastModifiedTimestamp updated. + */ + public static void scanChanges() { + MONITOR_INSTANCES.forEach(group -> { + try { + group.checkAndNotify(); + } catch (Throwable t) { + log.error("Files change detection failure, gourp = ", t); + } + }); + } + + /** + * Start the change monitoring. + */ + public void start() { + SCHEDULER_CHANGE_LOCK.lock(); + try { + if (FILE_MONITOR_TASK_SCHEDULER == null) { + FILE_MONITOR_TASK_SCHEDULER = Executors.newSingleThreadScheduledExecutor() + .scheduleAtFixedRate( + MultipleFilesChangeMonitor::scanChanges, 1, 200, + TimeUnit.MILLISECONDS + ); + } + + if (MONITOR_INSTANCES.contains(this)) { + throw new IllegalStateException("This FileChangeMonitor has been started."); + } + + this.checkAndNotify(); + MONITOR_INSTANCES.add(this); + } finally { + SCHEDULER_CHANGE_LOCK.unlock(); + } + } + + /** + * Stop the change monitoring. + */ + public void stop() { + SCHEDULER_CHANGE_LOCK.lock(); + try { + MONITOR_INSTANCES.remove(this); + } finally { + SCHEDULER_CHANGE_LOCK.unlock(); + } + } + + @Override + public String toString() { + return "MultipleFilesChangeMonitor{" + + "watchedFiles=" + watchedFiles + + '}'; + } + + /** + * The callback when files changed. + */ + public interface FilesChangedNotifier { + /** + * Notify the new content by providing the file input stream for all files in this group. + * + * @param readableContents include the new contents. NULL if the file doesn't exist. + */ + void filesChanged(List readableContents); + } + + /** + * WatchedFile represents a file change detector. It could detect the file changed based on modified time and file + * content at the binary level. It load the file content into the memory as cache to do the comparison. + */ + @RequiredArgsConstructor + @Slf4j + private static class WatchedFile { + /** + * The absolute path of the monitored file. + */ + private final String filePath; + /** + * The last modify time of the {@link #filePath} + */ + private long lastModifiedTimestamp = 0; + /** + * File content at the latest status. + */ + private byte[] fileContent; + + /** + * Detect the file content change, if yes, reload the file content into the memory as cached data. + * + * @return true if file content changed. + */ + boolean detectContentChanged() { + File targetFile = new File(filePath); + if (!targetFile.exists()) { + if (lastModifiedTimestamp == 0) { + //File doesn't exist before, no change detected. + return false; + } else { + // File has been deleted. Reset the modified timestamp. + lastModifiedTimestamp = 0; + return true; + } + } else { + long lastModified = targetFile.lastModified(); + if (lastModified != lastModifiedTimestamp) { + // File modified timestamp changed. Need to read the file content. + try (FileInputStream fileInputStream = new FileInputStream(targetFile)) { + byte[] b = new byte[1024]; + ByteArrayOutputStream os = new ByteArrayOutputStream(); + int c; + while ((c = fileInputStream.read(b)) != -1) { + os.write(b, 0, c); + } + byte[] newContent = os.toByteArray(); + if (!Arrays.equals(newContent, fileContent)) { + fileContent = newContent; + return true; + } else { + return false; + } + } catch (FileNotFoundException e) { + log.error("The existed file turns to missing, watch file=" + filePath, e); + } catch (IOException e) { + log.error("Read file failure, watch file=" + filePath, e); + } finally { + lastModifiedTimestamp = lastModified; + } + } + return false; + } + } + } +} diff --git a/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/MultipleFilesChangeMonitorTest.java b/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/MultipleFilesChangeMonitorTest.java new file mode 100644 index 000000000..6e8ecc734 --- /dev/null +++ b/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/MultipleFilesChangeMonitorTest.java @@ -0,0 +1,92 @@ +/* + * 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.skywalking.oap.server.library.util; + +import java.io.BufferedOutputStream; +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.nio.charset.Charset; +import java.util.List; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +public class MultipleFilesChangeMonitorTest { + private static String FILE_NAME = "FileChangeMonitorTest.tmp"; + + @Test + public void test() throws InterruptedException, IOException { + StringBuilder content = new StringBuilder(); + MultipleFilesChangeMonitor monitor = new MultipleFilesChangeMonitor( + 1, new MultipleFilesChangeMonitor.FilesChangedNotifier() { + + @Override + public void filesChanged(final List readableContents) { + Assert.assertEquals(2, readableContents.size()); + Assert.assertNull(readableContents.get(1)); + try { + content.delete(0, content.length()); + content.append(new String(readableContents.get(0), 0, readableContents.get(0).length, "UTF-8")); + } catch (UnsupportedEncodingException e) { + e.printStackTrace(); + } + } + }, FILE_NAME, "XXXX_NOT_EXIST.SW"); + + monitor.start(); + + File file = new File(FILE_NAME); + BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(file)); + bos.write("test context".getBytes(Charset.forName("UTF-8"))); + bos.flush(); + bos.close(); + + int countDown = 40; + boolean notified = false; + boolean notified2 = false; + while (countDown-- > 0) { + if ("test context".equals(content.toString())) { + file = new File(FILE_NAME); + bos = new BufferedOutputStream(new FileOutputStream(file)); + bos.write("test context again".getBytes(Charset.forName("UTF-8"))); + bos.flush(); + bos.close(); + notified = true; + } else if ("test context again".equals(content.toString())) { + notified2 = true; + break; + } + Thread.sleep(500); + } + Assert.assertTrue(notified); + Assert.assertTrue(notified2); + } + + @BeforeClass + @AfterClass + public static void cleanup() { + File file = new File(FILE_NAME); + if (file.exists() && file.isFile()) { + file.delete(); + } + } +}