[
https://issues.apache.org/jira/browse/GOBBLIN-1678?focusedWorklogId=802620&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-802620
]
ASF GitHub Bot logged work on GOBBLIN-1678:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 22/Aug/22 20:15
Start Date: 22/Aug/22 20:15
Worklog Time Spent: 10m
Work Description: arjun4084346 commented on code in PR #3536:
URL: https://github.com/apache/gobblin/pull/3536#discussion_r951871662
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GitFlowGraphMonitor.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.gobblin.service.monitoring;
+
+import com.google.common.collect.ImmutableMap;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.hadoop.fs.Path;
+import org.eclipse.jgit.api.errors.GitAPIException;
+import org.eclipse.jgit.diff.DiffEntry;
+
+import com.google.common.base.Optional;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.DataNode;
+import org.apache.gobblin.service.modules.flowgraph.FlowEdge;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import
org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+
+
+/**
+ * Service that monitors for changes to {@link
org.apache.gobblin.service.modules.flowgraph.FlowGraph} from a git repository.
+ * The git repository must have an inital commit that has no files since that
is used as a base for getting
+ * the change list.
+ * The {@link DataNode}s and {@link FlowEdge}s in FlowGraph need to be
organized with the following directory structure on git:
+ * <root_flowGraph_dir>/<nodeName>/<nodeName>.properties
+ * <root_flowGraph_dir>/<nodeName1>/<nodeName2>/<edgeName>.properties
+ */
+@Slf4j
+public class GitFlowGraphMonitor extends GitMonitoringService implements
FlowGraphMonitor {
+ public static final String GIT_FLOWGRAPH_MONITOR_PREFIX =
"gobblin.service.gitFlowGraphMonitor";
+ private static final String PROPERTIES_EXTENSIONS = "properties";
+ private static final String CONF_EXTENSIONS = "conf";
+ private static final String DEFAULT_GIT_FLOWGRAPH_MONITOR_REPO_DIR =
"git-flowgraph";
+ private static final String DEFAULT_GIT_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR =
"gobblin-flowgraph";
+ private static final String DEFAULT_GIT_FLOWGRAPH_MONITOR_BRANCH_NAME =
"master";
+ static final String SHOULD_CHECKPOINT_HASHES = "shouldCheckpointHashes";
+
+ private static final int DEFAULT_GIT_FLOWGRAPH_MONITOR_POLLING_INTERVAL = 60;
+
+ private static final Config DEFAULT_FALLBACK =
ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.GIT_MONITOR_REPO_DIR,
DEFAULT_GIT_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.GIT_MONITOR_CONFIG_BASE_DIR,
DEFAULT_GIT_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.GIT_MONITOR_BRANCH_NAME,
DEFAULT_GIT_FLOWGRAPH_MONITOR_BRANCH_NAME)
+ .put(ConfigurationKeys.GIT_MONITOR_POLLING_INTERVAL,
DEFAULT_GIT_FLOWGRAPH_MONITOR_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, CONF_EXTENSIONS)
+ .put(SHOULD_CHECKPOINT_HASHES, false)
+ .build());
+
+ private Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog;
+ private final CountDownLatch initComplete;
+
+ public GitFlowGraphMonitor(Config config, Optional<? extends
FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch
initComplete) {
+
super(config.getConfig(GIT_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK));
+ Config configWithFallbacks =
config.getConfig(GIT_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.flowTemplateCatalog = flowTemplateCatalog;
+ this.initComplete = initComplete;
+ this.listeners.add(new GitFlowGraphListener(
+ flowTemplateCatalog, graph, topologySpecMap,
configWithFallbacks.getString(ConfigurationKeys.GIT_MONITOR_REPO_DIR),
+
configWithFallbacks.getString(ConfigurationKeys.GIT_MONITOR_CONFIG_BASE_DIR),
configWithFallbacks.getString(ConfigurationKeys.JAVA_PROPS_EXTENSIONS),
+ configWithFallbacks.getString(ConfigurationKeys.HOCON_FILE_EXTENSIONS))
+ );
+ }
+
+ /**
+ * Determine if the service should poll Git. Current behavior is both
leaders and followers(s) will poll Git for
+ * changes to {@link FlowGraph}.
+ */
+ @Override
+ public boolean shouldPollGit() {
+ return this.isActive;
Review Comment:
code says it should poll git only when it is active, but the javadoc says
both the leader and followers should poll the git. Do the follower also have
isActive=true?
Issue Time Tracking
-------------------
Worklog Id: (was: 802620)
Time Spent: 2h 20m (was: 2h 10m)
> Refactor GaaS Flowgraph Monitor to be extensible
> ------------------------------------------------
>
> Key: GOBBLIN-1678
> URL: https://issues.apache.org/jira/browse/GOBBLIN-1678
> Project: Apache Gobblin
> Issue Type: Improvement
> Reporter: William Lo
> Priority: Major
> Time Spent: 2h 20m
> Remaining Estimate: 0h
>
> To support new implementations of a flow graph monitor, which allows for live
> updating of a flowgraph, we should reuse as much implementation from the
> existing git flowgraph monitor as possible.
> The current flowgraph monitor has coupled logic to perform a lot of the
> adding node/edges which can be reused for a file based flowgraph.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
