Provide a way to listen to the events of mirroring

Motivation: Mirroring failure is only recorded as a warning, but there is a need to handle it in a different way. For example, it can be recorded in metrics or end-users can be notified immediately. `MirrorListener` is provided as an extension point to utilize various events occurring in the mirror. Modifications: - Introduce `MirrorListener` whose implementations can be loaded dynamically via Java SPI. - `onStart()`, `onComplete()` and `onError()` events are added. - The default behavior is preserved in `DefaultMirrorListener` which is only used when no custom `MirrorListener` is configured. Result: You can now use `MirrorListener` to listen to `Mirror` events.
line · ikhoon · Nov 12, 2024 · Nov 8, 2024 · Nov 12, 2024 · Nov 11, 2024
commit d57496922bf322d82e85c1d3489abaaf7b39c1cb
@@ -0,0 +1,140 @@
+/*
+ * Copyright 2024 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://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 com.linecorp.centraldogma.it.mirror.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.net.URI;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import com.cronutils.model.Cron;
+import com.cronutils.model.CronType;
+import com.cronutils.model.definition.CronDefinitionBuilder;
+import com.cronutils.parser.CronParser;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import com.linecorp.centraldogma.server.command.CommandExecutor;
+import com.linecorp.centraldogma.server.credential.Credential;
+import com.linecorp.centraldogma.server.internal.mirror.AbstractMirror;
+import com.linecorp.centraldogma.server.internal.mirror.MirrorSchedulingService;
+import com.linecorp.centraldogma.server.mirror.Mirror;
+import com.linecorp.centraldogma.server.mirror.MirrorDirection;
+import com.linecorp.centraldogma.server.mirror.MirrorResult;
+import com.linecorp.centraldogma.server.mirror.MirrorStatus;
+import com.linecorp.centraldogma.server.storage.project.Project;
+import com.linecorp.centraldogma.server.storage.project.ProjectManager;
+import com.linecorp.centraldogma.server.storage.repository.MetaRepository;
+import com.linecorp.centraldogma.server.storage.repository.Repository;
+
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+
+class CustomMirrorListenerTest {
+
+    private static final Cron EVERY_SECOND = new CronParser(
+            CronDefinitionBuilder.instanceDefinitionFor(CronType.QUARTZ)).parse("* * * * * ?");
+
+    @TempDir
+    static File temporaryFolder;
+
+    @BeforeEach
+    void setUp() {
+        TestMirrorListener.reset();
+    }
+
+    @AfterEach
+    void tearDown() {
+        TestMirrorListener.reset();
+    }
+
+    @Test
+    void shouldNotifyMirrorEvents() {
+        final AtomicInteger taskCounter = new AtomicInteger();
+        final ProjectManager pm = mock(ProjectManager.class);
+        final Project p = mock(Project.class);
+        final MetaRepository mr = mock(MetaRepository.class);
+        final Repository r = mock(Repository.class);
+        when(pm.list()).thenReturn(ImmutableMap.of("foo", p));
+        when(p.name()).thenReturn("foo");
+        when(p.metaRepo()).thenReturn(mr);
+        when(r.parent()).thenReturn(p);
+        when(r.name()).thenReturn("bar");
+
+        final Mirror mirror = new AbstractMirror("my-mirror-1", true, EVERY_SECOND,
+                                                 MirrorDirection.REMOTE_TO_LOCAL,
+                                                 Credential.FALLBACK, r, "/",
+                                                 URI.create("unused://uri"), "/", "", null) {
+            @Override
+            protected MirrorResult mirrorLocalToRemote(File workDir, int maxNumFiles, long maxNumBytes,
+                                                       Instant triggeredTime) {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            protected MirrorResult mirrorRemoteToLocal(File workDir, CommandExecutor executor,
+                                                       int maxNumFiles, long maxNumBytes, Instant triggeredTime)
+                    throws Exception {
+                final int counter = taskCounter.incrementAndGet();
+                if (counter == 1) {
+                    return newMirrorResult(MirrorStatus.SUCCESS, "1", Instant.now());
+                } else if (counter == 2) {
+                    return newMirrorResult(MirrorStatus.UP_TO_DATE, "2", Instant.now());
+                } else {
+                    throw new IllegalStateException("failed");
+                }
+            }
+        };
+
+        when(mr.mirrors()).thenReturn(CompletableFuture.completedFuture(ImmutableList.of(mirror)));
+
+        final MirrorSchedulingService service = new MirrorSchedulingService(
+                temporaryFolder, pm, new SimpleMeterRegistry(), 1, 1, 1);
+        final CommandExecutor executor = mock(CommandExecutor.class);
+        service.start(executor);
+
+        try {
+            await().until(() -> taskCounter.get() >= 3);
+        } finally {
+            service.stop();
+        }
+        final Integer startCount = TestMirrorListener.startCount.get(mirror);
+        assertThat(startCount).isGreaterThanOrEqualTo(3);
+
+        final List<MirrorResult> completions = TestMirrorListener.completions.get(mirror);
+        assertThat(completions).hasSize(2);
+        assertThat(completions.get(0).mirrorStatus()).isEqualTo(MirrorStatus.SUCCESS);
+        assertThat(completions.get(1).mirrorStatus()).isEqualTo(MirrorStatus.UP_TO_DATE);
+
+        final List<Throwable> errors = TestMirrorListener.errors.get(mirror);
+        assertThat(errors).hasSizeGreaterThanOrEqualTo(1);
+        assertThat(errors.get(0).getCause())
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessage("failed");
+    }
+}
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2024 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://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 com.linecorp.centraldogma.it.mirror.listener;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.linecorp.centraldogma.server.mirror.Mirror;
+import com.linecorp.centraldogma.server.mirror.MirrorListener;
+import com.linecorp.centraldogma.server.mirror.MirrorResult;
+
+public final class TestMirrorListener implements MirrorListener {
+
+    static final Map<Mirror, Integer> startCount = new ConcurrentHashMap<>();
+    static final Map<Mirror, List<MirrorResult>> completions = new ConcurrentHashMap<>();
+    static final Map<Mirror, List<Throwable>> errors = new ConcurrentHashMap<>();
+
+    static void reset() {
+        startCount.clear();
+        completions.clear();
+        errors.clear();
+    }
+
+    @Override
+    public void onStart(Mirror mirror) {
+        startCount.merge(mirror, 1, Integer::sum);
+    }
+
+    @Override
+    public void onComplete(Mirror mirror, MirrorResult result) {
+        final ArrayList<MirrorResult> results = new ArrayList<>();
+        results.add(result);
+        completions.merge(mirror, results, (oldValue, newValue) -> {
+            oldValue.addAll(newValue);
+            return oldValue;
+        });
+    }
+
+    @Override
+    public void onError(Mirror mirror, Throwable cause) {
+        //noinspection ArraysAsListWithZeroOrOneArgument
+        errors.merge(mirror, Arrays.asList(cause), (oldValue, newValue) -> {
+            oldValue.addAll(newValue);
+            return oldValue;
+        });
+    }
+}
@@ -0,0 +1 @@
+com.linecorp.centraldogma.it.mirror.listener.TestMirrorListener
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2024 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://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 com.linecorp.centraldogma.server.internal.mirror;
+
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.linecorp.centraldogma.server.mirror.Mirror;
+import com.linecorp.centraldogma.server.mirror.MirrorListener;
+import com.linecorp.centraldogma.server.mirror.MirrorResult;
+
+final class CompositeMirrorListener implements MirrorListener {
+
+    private static final Logger logger = LoggerFactory.getLogger(CompositeMirrorListener.class);
+
+    private final List<MirrorListener> delegates;
+
+    CompositeMirrorListener(List<MirrorListener> delegates) {
+        this.delegates = delegates;
+    }
+
+    @Override
+    public void onStart(Mirror mirror) {
+        for (MirrorListener delegate : delegates) {
+            try {
+                delegate.onStart(mirror);
+            } catch (Exception e) {
+                logger.warn("Failed to notify a listener of the mirror start event: {}", delegate, e);
+            }
+        }
+    }
+
+    @Override
+    public void onComplete(Mirror mirror, MirrorResult result) {
+        for (MirrorListener delegate : delegates) {
+            try {
+                delegate.onComplete(mirror, result);
+            } catch (Exception e) {
+                logger.warn("Failed to notify a listener of the mirror complete event: {}", delegate, e);
+            }
+        }
+    }
+
+    @Override
+    public void onError(Mirror mirror, Throwable cause) {
+        for (MirrorListener delegate : delegates) {
+            try {
+                delegate.onError(mirror, cause);
+            } catch (Exception e) {
+                logger.warn("Failed to notify a listener of the mirror error event: {}", delegate, e);
+            }
+        }
+    }
+}
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2024 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://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 com.linecorp.centraldogma.server.internal.mirror;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.linecorp.centraldogma.server.mirror.Mirror;
+import com.linecorp.centraldogma.server.mirror.MirrorListener;
+import com.linecorp.centraldogma.server.mirror.MirrorResult;
+
+enum DefaultMirrorListener implements MirrorListener {
+
+    INSTANCE;
+
+    private static final Logger logger = LoggerFactory.getLogger(DefaultMirrorListener.class);
+
+    @Override
+    public void onStart(Mirror mirror) {
+        logger.info("Mirroring: {}", mirror);
+    }
+
+    @Override
+    public void onComplete(Mirror mirror, MirrorResult result) {
+        // Do nothing
+    }
+
+    @Override
+    public void onError(Mirror mirror, Throwable cause) {
+        logger.warn("Unexpected exception while mirroring: {}", mirror, cause);
+    }
+}
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		com.linecorp.centraldogma.it.mirror.listener.TestMirrorListener