You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cs...@apache.org on 2021/01/21 14:17:31 UTC
[sling-whiteboard] branch master updated: SLING-10067 - Added
monitoring
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-whiteboard.git
The following commit(s) were added to refs/heads/master by this push:
new b23a903 SLING-10067 - Added monitoring
b23a903 is described below
commit b23a90364d03dd6d0b9db2218400208ed4dea58e
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Thu Jan 21 15:17:15 2021 +0100
SLING-10067 - Added monitoring
---
chunked-distribution/README.md | 29 ++++++++++--
.../distribution/chunked/ChunkedDistribution.java | 52 +++++++++++++++-------
.../chunked/ChunkedDistributionServlet.java | 44 +++++++++++++++++-
.../sling/distribution/chunked/DeepTree.java | 22 ++++-----
.../apache/sling/distribution/chunked/Mode.java | 21 +++++++++
.../chunked/ChunkedDistributionTest.java | 2 +-
.../sling/distribution/chunked/DeepTreeTest.java | 2 +-
7 files changed, 138 insertions(+), 34 deletions(-)
diff --git a/chunked-distribution/README.md b/chunked-distribution/README.md
index 537e905..bc6ba2c 100644
--- a/chunked-distribution/README.md
+++ b/chunked-distribution/README.md
@@ -1,6 +1,5 @@
# chunked-distribution
-https://cwiki.apache.org/confluence/display/SLING/Service+Authentication
"org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended~chunked-distribution":{
"user.default":"",
@@ -10,6 +9,28 @@ https://cwiki.apache.org/confluence/display/SLING/Service+Authentication
},
- curl -u admin:admin -F action=install -F bundlefile=@"target/org.apache.sling.distribution.chunked-0.1.0-SNAPSHOT.jar" http://localhost:8080/system/console/bundles
- curl -u admin:admin -X POST -d "apply=true" -d "propertylist=user.mapping" -d "user.mapping=org.apache.sling.distribution.chunked=[repository-reader-service]" -d "factoryPid=org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended" http://localhost:8080/system/console/configMgr
-
\ No newline at end of file
+ curl -i -u admin:admin -F action=install -F bundlestart=true -F bundlefile=@"target/org.apache.sling.distribution.chunked-0.1.0-SNAPSHOT.jar" http://localhost:8080/system/console/bundles
+
+ curl -u admin:admin -X POST -d "apply=true" -d "propertylist=user.mapping" -d "user.mapping=org.apache.sling.distribution.chunked\=[repository-reader-service]" -d "factoryPid=org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended" http://localhost:8080/system/console/configMgr
+
+
+## Stop job
+
+ curl -i -u admin:admin -F command=stop -F id=<job id> http://localhost:8080/libs/sling/distribution/tree
+
+## Distribute path with just hierarchy noes
+
+ curl -i -u admin:admin -F path=<path> http://localhost:8080/libs/sling/distribution/tree
+
+## Distribute path with all nodes
+
+ curl -i -u admin:admin -F path=<path> -Fmode=AllNodes http://localhost:8080/libs/sling/distribution/tree
+
+
+## Get jobs
+
+ curl -i -u admin:admin http://localhost:8080/libs/sling/distribution/tree
+
+## Observe progress
+
+ watch -t -d curl -u admin:admin http://localhost:8080/libs/sling/distribution/tree
diff --git a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistribution.java b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistribution.java
index 4528938..ca6a996 100644
--- a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistribution.java
+++ b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistribution.java
@@ -42,8 +42,10 @@ import org.slf4j.LoggerFactory;
@Component(property = { JobConsumer.PROPERTY_TOPICS + "=" + ChunkedDistribution.TOPIC })
public class ChunkedDistribution implements JobExecutor {
- private static final int CHUNK_SIZE = 100;
-
+ public static final int DEFAULT_CHUNK_SIZE = 1000;
+ public static final String KEY_PATH = "path";
+ public static final String KEY_MODE = "mode";
+ public static final String KEY_CHUNK_SIZE = "chunkSize";
public static final String TOPIC = "sling/whiteboard/distribution/chunked";
private Logger log = LoggerFactory.getLogger(this.getClass());
@@ -61,53 +63,71 @@ public class ChunkedDistribution implements JobExecutor {
@Override
public JobExecutionResult process(Job job, JobExecutionContext context) {
try {
- String path = Objects.requireNonNull(job.getProperty("path", String.class), "No path parameter provided");
+ String path = requireParam(job, KEY_PATH, String.class);
+ String modeSt = requireParam(job, KEY_MODE, String.class);
+ Mode mode = Mode.valueOf(modeSt);
+ Integer chunkSize = requireParam(job, KEY_CHUNK_SIZE, Integer.class);
+ log.info("Starting chunked tree distribution for path {}", path);
try (ResourceResolver resolver = resolverFactory.getServiceResourceResolver(null)) {
- distribute(resolver, path, context);
+ distribute(resolver, path, mode, chunkSize, context);
+ log.info("Finished chunked tree distribution for path {}", path);
return context.result().succeeded();
}
} catch (Exception e) {
+ log.warn(e.getMessage(), e);
+ context.log(e.getMessage());
return context.result().message(e.getMessage()).cancelled();
}
}
- public void distribute(ResourceResolver resolver, String path, JobExecutionContext context) {
- Resource parent = Objects.requireNonNull(resolver.getResource(path), "No resource present at path " + path);
+ private <T> T requireParam(Job job, String key, Class<T> type) {
+ return Objects.requireNonNull(job.getProperty(key, type), "No " + key + " parameter provided");
+ }
- List<String> paths = DeepTree.getPaths(parent);
- List<List<String>> chunks = getChunks(paths);
+ public void distribute(ResourceResolver resolver, String path, Mode mode, Integer chunkSize, JobExecutionContext context) {
+ Resource parent = Objects.requireNonNull(resolver.getResource(path), "No resource present at path " + path);
+ context.log("Getting tree nodes for path=" + path);
+ List<String> paths = DeepTree.getPaths(parent, mode);
+ List<List<String>> chunks = getChunks(paths, chunkSize);
context.initProgress(chunks.size(), -1);
+ int progress = 0;
for (List<String> chunk : chunks) {
context.incrementProgressCount(1);
- distributeChunk(resolver, chunk);
+ progress ++;
+ String firstPath = chunk.iterator().next();
+ String msg = String.format("Distributing chunk %d/%d starting with %s", progress, chunks.size(), firstPath);
+ log.info(msg);
+ context.log(msg);
+ distributeChunk(resolver, chunk, context);
if (context.isStopped()) {
throw new RuntimeException("Job stopped");
}
}
}
- private List<List<String>> getChunks(List<String> paths) {
+ private List<List<String>> getChunks(List<String> paths, Integer chunkSize) {
List<List<String>> chunks = new ArrayList<>();
int c = 0;
while (c < paths.size()) {
- int next = Math.min(paths.size(), c + CHUNK_SIZE);
+ int next = Math.min(paths.size(), c + chunkSize);
chunks.add(paths.subList(c, next));
c = next;
}
return chunks;
}
- private void distributeChunk(ResourceResolver resolver, List<String> paths) {
- String firstPath = paths.iterator().next();
+ private void distributeChunk(ResourceResolver resolver, List<String> paths, JobExecutionContext context) {
try {
String[] pathsAr = paths.toArray(new String[] {});
- log.info("Distributing chunk starting with {}", firstPath);
DistributionRequest request = new SimpleDistributionRequest(DistributionRequestType.ADD, pathsAr);
distributor.distribute("publish", resolver, request);
- log.info("Distributing request created");
} catch (Exception e) {
- log.warn("Error creating distribution request", firstPath);
+ String firstPath = paths.iterator().next();
+ String msg = "Error creating distribution request first path " + firstPath + " msg: " + e.getMessage();
+ context.log(msg);
+ log.warn(msg, e);
}
}
+
}
diff --git a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistributionServlet.java b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistributionServlet.java
index efd5932..1d7fcd1 100644
--- a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistributionServlet.java
+++ b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/ChunkedDistributionServlet.java
@@ -19,8 +19,13 @@
package org.apache.sling.distribution.chunked;
import java.io.IOException;
+import java.io.PrintWriter;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import java.util.TimeZone;
import javax.servlet.Servlet;
import javax.servlet.ServletException;
@@ -28,7 +33,9 @@ import javax.servlet.ServletException;
import org.apache.sling.api.SlingHttpServletRequest;
import org.apache.sling.api.SlingHttpServletResponse;
import org.apache.sling.api.servlets.SlingAllMethodsServlet;
+import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.JobManager;
+import org.apache.sling.event.jobs.JobManager.QueryType;
import org.apache.sling.servlets.annotations.SlingServletPaths;
import org.apache.sling.servlets.annotations.SlingServletResourceTypes;
import org.osgi.service.component.annotations.Activate;
@@ -56,15 +63,48 @@ public class ChunkedDistributionServlet extends SlingAllMethodsServlet {
@Override
protected void doGet(SlingHttpServletRequest request, SlingHttpServletResponse response)
throws ServletException, IOException {
+ TimeZone tz = TimeZone.getTimeZone("UTC");
+ DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm'Z'"); // Quoted "Z" to indicate UTC, no timezone offset
+ df.setTimeZone(tz);
+ PrintWriter wr = response.getWriter();
+ Collection<Job> jobs = jobMananger.findJobs(QueryType.ACTIVE, ChunkedDistribution.TOPIC, -1, (Map<String, Object>[]) null);
+ wr.println("Jobs");
+ for (Job job : jobs) {
+ String startedAt = df.format(job.getProcessingStarted().getTime());
+ wr.println(String.format("id: %s, step: %d/%d, startedAt: %s", job.getId(), job.getFinishedProgressStep(), job.getProgressStepCount(), startedAt));
+ String[] log = job.getProgressLog();
+ if (log != null) {
+ for (String line : log) {
+ wr.println(" " + line);
+ }
+ }
+ }
}
@Override
protected void doPost(SlingHttpServletRequest request, SlingHttpServletResponse response)
throws ServletException, IOException {
- String path = request.getParameter("path");
+ PrintWriter wr = response.getWriter();
+ String command = request.getParameter("command");
+ String jobId = request.getParameter("id");
+ if ("stop".equals(command)) {
+ jobMananger.stopJobById(jobId);
+ wr.println("Sent stop signal to " + jobId);
+ return;
+ }
+ String path = request.getParameter(ChunkedDistribution.KEY_PATH);
+ String modeSt = request.getParameter(ChunkedDistribution.KEY_MODE);
+ String mode = modeSt != null ? modeSt : Mode.OnlyHierarchyNodes.name();
+
+ String chunkSizeSt = request.getParameter(ChunkedDistribution.KEY_CHUNK_SIZE);
+ Integer chunkSize = chunkSizeSt != null ? Integer.parseInt(chunkSizeSt) : ChunkedDistribution.DEFAULT_CHUNK_SIZE;
+
Map<String, Object> props = new HashMap<>();
props.put("path", path);
- jobMananger.addJob(ChunkedDistribution.TOPIC, props);
+ props.put("mode", mode);
+ props.put("chunkSize", chunkSize);
+ Job job = jobMananger.addJob(ChunkedDistribution.TOPIC, props);
+ wr.println(job.getId());
}
}
diff --git a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/DeepTree.java b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/DeepTree.java
index 8eee9d7..451057b 100644
--- a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/DeepTree.java
+++ b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/DeepTree.java
@@ -32,25 +32,28 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DeepTree {
- private Logger log = LoggerFactory.getLogger(this.getClass());
+ private static Logger log = LoggerFactory.getLogger(DeepTree.class);
private List<String> paths = new ArrayList<>();
+ private Mode mode;
- public static List<String> getPaths(Resource baseResource) {
- DeepTree walker = new DeepTree(baseResource);
+ public static List<String> getPaths(Resource baseResource, Mode mode) {
+ String path = Objects.requireNonNull(baseResource).getPath();
+ log.info("Getting deep tree for {} using mode {}", path, mode);
+ DeepTree walker = new DeepTree(mode);
+ walker.walkTreeRecursively(baseResource);
+ log.info("Getting deep tree for {} finished with {} results", path, walker.paths.size());
return walker.paths;
}
- private DeepTree(Resource baseResource) {
- log.info("Getting deep tree for {}", Objects.requireNonNull(baseResource).getPath());
- walkTreeRecursively(baseResource);
- log.info("Getting deep tree finished with {} results", paths.size());
+ private DeepTree(Mode mode) {
+ this.mode = mode;
}
public void walkTreeRecursively(Resource baseResource) {
try {
Node baseNode = baseResource.adaptTo(Node.class);
boolean isHierarchyNode = baseNode.isNodeType("nt:hierarchyNode");
- if (isHierarchyNode) {
+ if (mode == Mode.AllNodes || isHierarchyNode) {
paths.add(baseResource.getPath());
Iterable<Resource> childrenIter = baseResource.getChildren();
List<Resource> children = getChildren(childrenIter.iterator());
@@ -70,6 +73,5 @@ public class DeepTree {
children.add(childrenIt.next());
}
return children;
- }
-
+ };
}
diff --git a/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/Mode.java b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/Mode.java
new file mode 100644
index 0000000..069adc1
--- /dev/null
+++ b/chunked-distribution/src/main/java/org/apache/sling/distribution/chunked/Mode.java
@@ -0,0 +1,21 @@
+/*
+ * 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.sling.distribution.chunked;
+
+public enum Mode { AllNodes, OnlyHierarchyNodes }
\ No newline at end of file
diff --git a/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/ChunkedDistributionTest.java b/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/ChunkedDistributionTest.java
index cf31aa0..ccbeb28 100644
--- a/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/ChunkedDistributionTest.java
+++ b/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/ChunkedDistributionTest.java
@@ -75,7 +75,7 @@ public class ChunkedDistributionTest {
when(job.getProperty("path", String.class)).thenReturn("/test");
when(distributor.distribute(Mockito.eq("publish"), Mockito.any(ResourceResolver.class), requestCaptor.capture())).thenReturn(resp);
- dist.distribute(resolver, "/test", jcontext);
+ dist.distribute(resolver, "/test", Mode.OnlyHierarchyNodes, ChunkedDistribution.DEFAULT_CHUNK_SIZE, jcontext);
DistributionRequest request = requestCaptor.getValue();
diff --git a/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/DeepTreeTest.java b/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/DeepTreeTest.java
index 2e5f9e6..4b222c4 100644
--- a/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/DeepTreeTest.java
+++ b/chunked-distribution/src/test/java/org/apache/sling/distribution/chunked/DeepTreeTest.java
@@ -47,7 +47,7 @@ public class DeepTreeTest {
}
}
- List<String> paths = DeepTree.getPaths(base);
+ List<String> paths = DeepTree.getPaths(base, Mode.OnlyHierarchyNodes);
assertThat(paths.size(), Matchers.equalTo(100 + 10 + 1));
}
}