You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2022/09/01 17:19:22 UTC
[GitHub] [gobblin] Will-Lo opened a new pull request, #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Will-Lo opened a new pull request, #3548:
URL: https://github.com/apache/gobblin/pull/3548
… filesystem and updates dynamically
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1696
### Description
- [x] Here are some details about my PR, including screenshots (if applicable):
Gobblin-as-a-Service only has a Git based flowgraph, which is difficult to build CI/CD around.
This is an alternate flowgraph that watches for file changes and applies updates.
Unlike the git based flowgraph, changes in this flowgraph are atomic with respect to one deployment of the flowgraph, which is an update of multiple files.
This also requires an enhancement to the PathAlterationObserver to notify listeners if there is any change on a root level.
### Tests
- [x] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [x] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] ZihanLi58 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r974655361
##########
gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java:
##########
@@ -1022,8 +1022,14 @@ public class ConfigurationKeys {
* Configuration properties related to flowGraphs
*/
- public static final String JAVA_PROPS_EXTENSIONS = "javaPropsExtensions";
- public static final String HOCON_FILE_EXTENSIONS = "hoconFileExtensions";
+ public static final String JAVA_PROPS_EXTENSIONS = "flowGraph.javaPropsExtensions";
Review Comment:
Maybe include flowGraph in the config name as well?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -197,10 +200,9 @@ private boolean checkFilePath(String file, int depth) {
Path filePath = new Path(file);
Review Comment:
According to the comments on line 193, file is the relative path. But according to the code, seems we expect it to be absolute path. Can we update to make sure javadoc reflect the correct expectation?
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/ExceptionCatchingPathAlterationListenerDecorator.java:
##########
@@ -42,73 +45,83 @@ public Object getDecoratedObject() {
@Override
public void onStart(PathAlterationObserver observer) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onStart(observer);
- } catch (Throwable exc) {
- log.error("onStart failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileCreate(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileCreate(path);
- } catch (Throwable exc) {
- log.error("onFileCreate failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileChange(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileChange(path);
- } catch (Throwable exc) {
- log.error("onFileChange failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onStop(PathAlterationObserver observer) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onStop(observer);
- } catch (Throwable exc) {
- log.error("onStop failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryCreate(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryCreate(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryCreate failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryChange(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryChange(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryChange failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryDelete(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryDelete(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryDelete failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileDelete(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileDelete(path);
+ return null;
+ });
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ logSwallowedThrowable(() -> {
+ this.underlying.onCheckDetectedChange();
+ return null;
+ });
+ }
+
+ protected void logSwallowedThrowable(Callable<Void> c) {
+ try {
+ c.call();
} catch (Throwable exc) {
- log.error("onFileDelete failure: ", exc);
+ String methodName = Arrays.stream(exc.getStackTrace()).findFirst().get().getMethodName();
+ log.error(methodName + " failed due to exception:", exc);
Review Comment:
Shouldn't we emit some metrics to indicate this failure? I feel this might ends up with silent failure and we cannot catch it immediately
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flow/MultiHopFlowCompiler.java:
##########
@@ -121,7 +124,7 @@ public MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrum
MultiHopFlowCompiler.log.warn("Exception reading data node alias map, ignoring it.", e);
}
- this.flowGraph = new BaseFlowGraph(dataNodeAliasMap);
+ this.flowGraph = new AtomicReference<>(new BaseFlowGraph(dataNodeAliasMap));
Review Comment:
Can we add comments to explain why we want use atomicReference here?
##########
gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java:
##########
@@ -1022,8 +1022,14 @@ public class ConfigurationKeys {
* Configuration properties related to flowGraphs
*/
- public static final String JAVA_PROPS_EXTENSIONS = "javaPropsExtensions";
- public static final String HOCON_FILE_EXTENSIONS = "hoconFileExtensions";
+ public static final String JAVA_PROPS_EXTENSIONS = "flowGraph.javaPropsExtensions";
+ public static final String HOCON_FILE_EXTENSIONS = "flowGraph.hoconFileExtensions";
Review Comment:
Same as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] AndyJiang99 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
AndyJiang99 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r975719456
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -84,26 +84,27 @@ public BaseFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempl
this.hoconFileExtensions = Sets.newHashSet(hoconFileExtensions.split(","));
try {
this.pullFileLoader = new PullFileLoader(folderPath,
- FileSystem.get(URI.create(ConfigurationKeys.LOCAL_FS_URI), new Configuration()),
- this.javaPropsExtensions, this.hoconFileExtensions);
+ FileSystem.get(URI.create(ConfigurationKeys.LOCAL_FS_URI), new Configuration()), this.javaPropsExtensions,
+ this.hoconFileExtensions);
} catch (IOException e) {
throw new RuntimeException("Could not create pull file loader", e);
}
}
+
/**
* Add a {@link DataNode} to the {@link FlowGraph}. The method uses the {@link FlowGraphConfigurationKeys#DATA_NODE_CLASS} config
* to instantiate a {@link DataNode} from the node config file.
* @param path of node to add
*/
- protected void addDataNode(String path) {
+ public void addDataNode(FlowGraph graph, String path) {
Review Comment:
Nit: indentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] AndyJiang99 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
AndyJiang99 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r975809338
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraphHelper;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static final long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private volatile boolean isActive = false;
+ private final long pollingInterval;
+ private BaseFlowGraphHelper flowGraphHelper;
+ private final PathAlterationObserverScheduler pathAlterationDetector;
+ private final FSPathAlterationFlowGraphListener listener;
+ private final PathAlterationObserver observer;
+ private final Path flowGraphPath;
+ private final AtomicReference<FlowGraph> flowGraph;
+ private final CountDownLatch initComplete;
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS).build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ AtomicReference<FlowGraph> graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete)
+ throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval =
+ TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
+ this.flowGraphHelper = new BaseFlowGraphHelper(flowTemplateCatalog, topologySpecMap, flowGraphPath.toString(),
+ configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_BASE_DIR), configWithFallbacks.getString(ConfigurationKeys.JAVA_PROPS_EXTENSIONS),
+ configWithFallbacks.getString(ConfigurationKeys.HOCON_FILE_EXTENSIONS));
+ this.listener = new FSPathAlterationFlowGraphListener(flowTemplateCatalog, graph, flowGraphPath.toString(), this.flowGraphHelper);
+
+ this.flowGraph = graph;
+ this.initComplete = initComplete;
+
+ if (pollingInterval == ConfigurationKeys.DISABLED_JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL) {
+ this.pathAlterationDetector = null;
+ } else {
+ this.pathAlterationDetector = new PathAlterationObserverScheduler(pollingInterval);
+ Optional<PathAlterationObserver> observerOptional = Optional.fromNullable(observer);
+ this.pathAlterationDetector.addPathAlterationObserver(this.listener, observerOptional,
+ this.flowGraphPath);
+ }
+ }
+
+ @Override
+ protected void startUp()
+ throws IOException {
+ }
+
+ @Override
+ public synchronized void setActive(boolean isActive) {
+ log.info("Setting the flow graph monitor to be " + isActive + " from " + this.isActive);
+ if (this.isActive == isActive) {
+ // No-op if already in correct state
+ return;
+ } else if (isActive) {
+ if (this.pathAlterationDetector != null) {
+ log.info("Starting the " + getClass().getSimpleName());
+ log.info("Polling flowgraph folder with interval {} ", this.pollingInterval);
+ try {
+ this.pathAlterationDetector.start();
+ // Manually instantiate flowgraph when the monitor becomes active
+ this.flowGraphHelper.populateFlowGraphAtomically(this.flowGraph);
+ // Reduce the countdown latch
+ this.initComplete.countDown();
+ log.info("Finished populating FSFlowgraph");
+ } catch (IOException e) {
+ log.error("Could not initialize pathAlterationDetector due to error: ", e);
+ }
+ } else {
+ log.warn("No path alteration detector found");
+ }
+ }
+ this.isActive = isActive;
Review Comment:
Should this be moved into the else-if statement instead? Since no-op if already in correct state
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970179511
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private boolean isActive;
+ private long pollingInterval;
+ private PathAlterationObserverScheduler pathAlterationDetector;
+ private PathAlterationObserver observer;
+ private Path flowGraphPath;
+
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS)
+ .build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete) throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval = TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
Review Comment:
It's actually used more as a ROOT_DIR directory, but it was hard to come up with naming that would not cause the git flowgraph configurations to break, if I were to share the same configuration keys. Maybe it's just better to define a new set of config keys for this flowgraph to improve the naming convention?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] codecov-commenter commented on pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#issuecomment-1234580553
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3548?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3548](https://codecov.io/gh/apache/gobblin/pull/3548?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (400b12f) into [master](https://codecov.io/gh/apache/gobblin/commit/d90d1961bbb8242a45266ec02eb0f27da5735c14?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (d90d196) will **decrease** coverage by `3.32%`.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## master #3548 +/- ##
============================================
- Coverage 46.64% 43.32% -3.33%
+ Complexity 10460 2036 -8424
============================================
Files 2087 406 -1681
Lines 81670 17541 -64129
Branches 9106 2145 -6961
============================================
- Hits 38095 7599 -30496
+ Misses 40055 9093 -30962
+ Partials 3520 849 -2671
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3548?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...eptionCatchingPathAlterationListenerDecorator.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvZmlsZXN5c3RlbS9FeGNlcHRpb25DYXRjaGluZ1BhdGhBbHRlcmF0aW9uTGlzdGVuZXJEZWNvcmF0b3IuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...util/filesystem/PathAlterationListenerAdaptor.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvZmlsZXN5c3RlbS9QYXRoQWx0ZXJhdGlvbkxpc3RlbmVyQWRhcHRvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...obblin/util/filesystem/PathAlterationObserver.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvZmlsZXN5c3RlbS9QYXRoQWx0ZXJhdGlvbk9ic2VydmVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.54% <0.00%> (-0.65%)` | :arrow_down: |
| [...pache/gobblin/configuration/ConfigurationKeys.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vY29uZmlndXJhdGlvbi9Db25maWd1cmF0aW9uS2V5cy5qYXZh) | | |
| [...bblin/service/modules/flowgraph/BaseFlowGraph.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9mbG93Z3JhcGgvQmFzZUZsb3dHcmFwaC5qYXZh) | | |
| [...rvice/modules/flowgraph/BaseFlowGraphListener.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9mbG93Z3JhcGgvQmFzZUZsb3dHcmFwaExpc3RlbmVyLmphdmE=) | | |
| [...ervice/modules/flowgraph/GitFlowGraphListener.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9mbG93Z3JhcGgvR2l0Rmxvd0dyYXBoTGlzdGVuZXIuamF2YQ==) | | |
| [...obblin/service/monitoring/GitFlowGraphMonitor.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9uaXRvcmluZy9HaXRGbG93R3JhcGhNb25pdG9yLmphdmE=) | | |
| [...etention/HivePartitionVersionRetentionCleaner.java](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tY29tcGxpYW5jZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9jb21wbGlhbmNlL3JldGVudGlvbi9IaXZlUGFydGl0aW9uVmVyc2lvblJldGVudGlvbkNsZWFuZXIuamF2YQ==) | | |
| ... and [1676 more](https://codecov.io/gh/apache/gobblin/pull/3548/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r966520982
##########
gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java:
##########
@@ -1020,8 +1020,14 @@ public class ConfigurationKeys {
* Configuration properties related to flowGraphs
*/
- public static final String JAVA_PROPS_EXTENSIONS = "javaPropsExtensions";
- public static final String HOCON_FILE_EXTENSIONS = "hoconFileExtensions";
+ public static final String JAVA_PROPS_EXTENSIONS = "flowGraph.javaPropsExtensions";
+ public static final String HOCON_FILE_EXTENSIONS = "flowGraph.hoconFileExtensions";
Review Comment:
Yes it does seem like they need to be, they use different pullers under the hood so this is needed to overwrite their differentiations. I'll add comments though. I think in the future we'd wan to move most of the flowgraph if not all of it to hocon format just because it's richer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r978337541
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/GitFlowGraphListener.java:
##########
@@ -53,10 +47,10 @@ public GitFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempla
@Override
public void addChange(DiffEntry change) {
Path path = new Path(change.getNewPath());
- if (path.depth() == NODE_FILE_DEPTH) {
- addDataNode(change.getNewPath());
- } else if (path.depth() == EDGE_FILE_DEPTH) {
- addFlowEdge(change.getNewPath());
+ if (path.depth() == BaseFlowGraphHelper.NODE_FILE_DEPTH) {
+ this.baseFlowGraphHelper.addDataNode(this.flowGraph.get(), change.getNewPath());
Review Comment:
I'm going to lean the latter approach which seems simpler to implement. Reason being is that I don't think that it is possible for the Git Flowgraph to run into issues with concurrent updates given the use of it (though it is theoretically possible) and also due to that this entire module will be deprecated soon anyways due to the FSFlowgraph.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r977047491
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraphHelper;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static final long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private volatile boolean isActive = false;
+ private final long pollingInterval;
+ private BaseFlowGraphHelper flowGraphHelper;
+ private final PathAlterationObserverScheduler pathAlterationDetector;
+ private final FSPathAlterationFlowGraphListener listener;
+ private final PathAlterationObserver observer;
+ private final Path flowGraphPath;
+ private final AtomicReference<FlowGraph> flowGraph;
+ private final CountDownLatch initComplete;
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS).build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ AtomicReference<FlowGraph> graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete)
+ throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval =
+ TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
+ this.flowGraphHelper = new BaseFlowGraphHelper(flowTemplateCatalog, topologySpecMap, flowGraphPath.toString(),
+ configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_BASE_DIR), configWithFallbacks.getString(ConfigurationKeys.JAVA_PROPS_EXTENSIONS),
+ configWithFallbacks.getString(ConfigurationKeys.HOCON_FILE_EXTENSIONS));
+ this.listener = new FSPathAlterationFlowGraphListener(flowTemplateCatalog, graph, flowGraphPath.toString(), this.flowGraphHelper);
+
+ this.flowGraph = graph;
+ this.initComplete = initComplete;
+
+ if (pollingInterval == ConfigurationKeys.DISABLED_JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL) {
+ this.pathAlterationDetector = null;
+ } else {
+ this.pathAlterationDetector = new PathAlterationObserverScheduler(pollingInterval);
+ Optional<PathAlterationObserver> observerOptional = Optional.fromNullable(observer);
+ this.pathAlterationDetector.addPathAlterationObserver(this.listener, observerOptional,
+ this.flowGraphPath);
+ }
+ }
+
+ @Override
+ protected void startUp()
+ throws IOException {
+ }
+
+ @Override
+ public synchronized void setActive(boolean isActive) {
+ log.info("Setting the flow graph monitor to be " + isActive + " from " + this.isActive);
+ if (this.isActive == isActive) {
+ // No-op if already in correct state
+ return;
+ } else if (isActive) {
+ if (this.pathAlterationDetector != null) {
+ log.info("Starting the " + getClass().getSimpleName());
+ log.info("Polling flowgraph folder with interval {} ", this.pollingInterval);
+ try {
+ this.pathAlterationDetector.start();
+ // Manually instantiate flowgraph when the monitor becomes active
+ this.flowGraphHelper.populateFlowGraphAtomically(this.flowGraph);
+ // Reduce the countdown latch
+ this.initComplete.countDown();
+ log.info("Finished populating FSFlowgraph");
+ } catch (IOException e) {
+ log.error("Could not initialize pathAlterationDetector due to error: ", e);
+ }
+ } else {
+ log.warn("No path alteration detector found");
+ }
+ }
+ this.isActive = isActive;
Review Comment:
And the noop does the early return
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970091256
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
+
+ private File graphDir;
+ CountDownLatch initComplete;
+
+ public FSPathAlterationFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, String baseDirectory, String flowGraphFolderName,
+ String javaPropsExtentions, String hoconFileExtensions, CountDownLatch initComplete) {
+ super(flowTemplateCatalog, graph, topologySpecMap, baseDirectory, flowGraphFolderName, javaPropsExtentions, hoconFileExtensions);
+ this.graphDir = new File(baseDirectory);
+ this.initComplete = initComplete;
+ // Populate the flowgraph with any existing files
+ if (!this.graphDir.exists()) {
+ throw new RuntimeException(String.format("Flowgraph directory at path %s does not exist!", graphDir));
+ }
+ this.populateFlowGraphAtomically();
+ }
+
+ public void onStart(final PathAlterationObserver observer) {
+ }
+
+ public void onFileCreate(final Path path) {
+ }
+
+ public void onFileChange(final Path path) {
+ }
+
+ public void onStop(final PathAlterationObserver observer) {
+ }
+
+ public void onDirectoryCreate(final Path directory) {
+ }
+
+ public void onDirectoryChange(final Path directory) {
+ }
+
+ public void onDirectoryDelete(final Path directory) {
+ }
+
+ public void onFileDelete(final Path path) {
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ log.info("Detecting change in flowgraph files, reloading flowgraph");
+ this.populateFlowGraphAtomically();
+ }
+
+ private void populateFlowGraphAtomically() {
+ FlowGraph newFlowGraph = new BaseFlowGraph();
+ try {
+ List<Path> edges = new ArrayList<>();
+ // All nodes must be added first before edges, otherwise edges may have a missing source or destination.
+ Files.walk(this.graphDir.toPath()).forEach(fileName -> {
+ if (!Files.isDirectory(fileName)) {
+ if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), NODE_FILE_DEPTH)) {
+ addDataNode(newFlowGraph, fileName.toString());
+ } else if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), EDGE_FILE_DEPTH)) {
+ edges.add(new Path(fileName.toString()));
+ }
+ }
+ });
+ for (Path edge: edges) {
+ addFlowEdge(newFlowGraph, edge.toString());
+ }
+ // Reduce the countdown latch
+ this.initComplete.countDown();
+ this.flowGraph.copyGraph(newFlowGraph);
Review Comment:
I'll try that, thanks :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r977046288
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/ExceptionCatchingPathAlterationListenerDecorator.java:
##########
@@ -42,73 +45,83 @@ public Object getDecoratedObject() {
@Override
public void onStart(PathAlterationObserver observer) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onStart(observer);
- } catch (Throwable exc) {
- log.error("onStart failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileCreate(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileCreate(path);
- } catch (Throwable exc) {
- log.error("onFileCreate failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileChange(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileChange(path);
- } catch (Throwable exc) {
- log.error("onFileChange failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onStop(PathAlterationObserver observer) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onStop(observer);
- } catch (Throwable exc) {
- log.error("onStop failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryCreate(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryCreate(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryCreate failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryChange(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryChange(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryChange failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryDelete(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryDelete(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryDelete failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileDelete(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileDelete(path);
+ return null;
+ });
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ logSwallowedThrowable(() -> {
+ this.underlying.onCheckDetectedChange();
+ return null;
+ });
+ }
+
+ protected void logSwallowedThrowable(Callable<Void> c) {
Review Comment:
It's representing a generic callable so I think a short name is fine here, usually we want more descriptive names for contextual variables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r979582884
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flow/MultiHopFlowCompiler.java:
##########
@@ -168,7 +171,7 @@ public MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrum
}
@VisibleForTesting
- MultiHopFlowCompiler(Config config, FlowGraph flowGraph) {
+ MultiHopFlowCompiler(Config config, AtomicReference<FlowGraph> flowGraph) {
Review Comment:
This is needed for testing, not sure how to be able to test the flow compiler otherwise without a getter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r979700601
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/GitFlowGraphListener.java:
##########
@@ -53,10 +47,10 @@ public GitFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempla
@Override
public void addChange(DiffEntry change) {
Path path = new Path(change.getNewPath());
- if (path.depth() == NODE_FILE_DEPTH) {
- addDataNode(change.getNewPath());
- } else if (path.depth() == EDGE_FILE_DEPTH) {
- addFlowEdge(change.getNewPath());
+ if (path.depth() == BaseFlowGraphHelper.NODE_FILE_DEPTH) {
+ this.baseFlowGraphHelper.addDataNode(this.flowGraph.get(), change.getNewPath());
Review Comment:
Nevermind I just went with making the gitflowgraphmonitor also update once per detected change in Git, seems like the more sensical solution given the complexity with the other approaches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] phet commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
phet commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r965504189
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraph.java:
##########
@@ -237,6 +237,24 @@ public Set<FlowEdge> getEdges(DataNode node) {
}
}
+ @Override
+ public void copyGraph(FlowGraph graph) {
Review Comment:
nit: this feels less making a copy (of the current instance), but overwriting the current instance with another. so maybe name `overwriteWith`.
thus renamed, it calls attention to the shared mutable state... which I suggest reconsidering, or at least clearly justifying in a comment (alternative sketched below)
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphListener.java:
##########
@@ -95,15 +95,15 @@ public BaseFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempl
* to instantiate a {@link DataNode} from the node config file.
* @param path of node to add
*/
- protected void addDataNode(String path) {
+ protected void addDataNode(FlowGraph graph, String path) {
Review Comment:
I see there's still a `this.flowGraph` field in this object. when would that differ from the `graph` param and/or would it ever be the same?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraph.java:
##########
@@ -237,6 +237,24 @@ public Set<FlowEdge> getEdges(DataNode node) {
}
}
+ @Override
+ public void copyGraph(FlowGraph graph) {
+ if (graph instanceof BaseFlowGraph) {
+ BaseFlowGraph baseFlowGraph = (BaseFlowGraph) graph;
+ try {
+ rwLock.writeLock().lock();
+ this.dataNodeAliasMap = baseFlowGraph.dataNodeAliasMap;
+ this.flowEdgeMap = baseFlowGraph.flowEdgeMap;
+ this.dataNodeMap = baseFlowGraph.dataNodeMap;
+ this.nodesToEdges = baseFlowGraph.nodesToEdges;
+ } finally {
+ rwLock.writeLock().unlock();
+ }
+ } else {
+ throw new UnsupportedOperationException("BaseFlowGraph can only clone other instances of BaseFlowGraph");
Review Comment:
suggest error to mention which class it was, when not `BaseFlowGraph`
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/PathAlterationListener.java:
##########
@@ -43,4 +43,6 @@ public interface PathAlterationListener {
void onDirectoryDelete(final Path directory);
void onFileDelete(final Path path);
+
+ void onCheckDetectedChange();
Review Comment:
javadoc would clarify... it's not immediately intuitive what "check detected change" is, in the way that "directory delete" is familiar.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
+
+ private File graphDir;
+ CountDownLatch initComplete;
+
+ public FSPathAlterationFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, String baseDirectory, String flowGraphFolderName,
+ String javaPropsExtentions, String hoconFileExtensions, CountDownLatch initComplete) {
+ super(flowTemplateCatalog, graph, topologySpecMap, baseDirectory, flowGraphFolderName, javaPropsExtentions, hoconFileExtensions);
+ this.graphDir = new File(baseDirectory);
+ this.initComplete = initComplete;
+ // Populate the flowgraph with any existing files
+ if (!this.graphDir.exists()) {
+ throw new RuntimeException(String.format("Flowgraph directory at path %s does not exist!", graphDir));
+ }
+ this.populateFlowGraphAtomically();
+ }
+
+ public void onStart(final PathAlterationObserver observer) {
+ }
+
+ public void onFileCreate(final Path path) {
+ }
+
+ public void onFileChange(final Path path) {
+ }
+
+ public void onStop(final PathAlterationObserver observer) {
+ }
+
+ public void onDirectoryCreate(final Path directory) {
+ }
+
+ public void onDirectoryChange(final Path directory) {
+ }
+
+ public void onDirectoryDelete(final Path directory) {
+ }
+
+ public void onFileDelete(final Path path) {
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ log.info("Detecting change in flowgraph files, reloading flowgraph");
+ this.populateFlowGraphAtomically();
+ }
+
+ private void populateFlowGraphAtomically() {
+ FlowGraph newFlowGraph = new BaseFlowGraph();
+ try {
+ List<Path> edges = new ArrayList<>();
+ // All nodes must be added first before edges, otherwise edges may have a missing source or destination.
+ Files.walk(this.graphDir.toPath()).forEach(fileName -> {
+ if (!Files.isDirectory(fileName)) {
+ if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), NODE_FILE_DEPTH)) {
+ addDataNode(newFlowGraph, fileName.toString());
+ } else if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), EDGE_FILE_DEPTH)) {
+ edges.add(new Path(fileName.toString()));
+ }
+ }
+ });
+ for (Path edge: edges) {
+ addFlowEdge(newFlowGraph, edge.toString());
Review Comment:
why turn a string into a `Path` only to convert it back `toString()` the only place it's used? if a reason clarify in a comment.
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/PathAlterationObserver.java:
##########
@@ -187,6 +186,14 @@ public void checkAndNotify()
for (final PathAlterationListener listener : listeners.values()) {
listener.onStop(this);
}
+
+ if (this.changeApplied) {
+ for (final PathAlterationListener listener : listeners.values()) {
+ // Fire onCheckDetectedChange to notify when one check contains any number of changes
+ listener.onCheckDetectedChange();
+ }
+ }
+ this.changeApplied = false;
Review Comment:
unless a semantic difference, I suggest to init to false as the first line of the method, rather than the last (which I believe to be leaving it for the next subsequent invocation).
##########
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitorTest.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Optional;
+import com.google.common.io.Files;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import org.apache.commons.io.FileUtils;
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.DataNode;
+import org.apache.gobblin.service.modules.flowgraph.FlowEdge;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphConfigurationKeys;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.hadoop.fs.Path;
+import org.eclipse.jgit.transport.RefSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class FsFlowGraphMonitorTest {
+ private static final Logger logger = LoggerFactory.getLogger(GitFlowGraphMonitor.class);
+ private final File TEST_DIR = new File(FileUtils.getTempDirectory(), "fsFlowGraphTestDir");
+ private final File flowGraphDir = new File(TEST_DIR, "/gobblin-flowgraph");
+ private static final String NODE_1_FILE = "node1.properties";
+ private final File node1Dir = new File(FileUtils.getTempDirectory(), "node1");
+ private final File node1File = new File(node1Dir, NODE_1_FILE);
+ private static final String NODE_2_FILE = "node2.properties";
+ private final File node2Dir = new File(FileUtils.getTempDirectory(), "node2");
+ private final File node2File = new File(node2Dir, NODE_2_FILE);
+ private final File edge1Dir = new File(node1Dir, "node2");
+ private final File edge1File = new File(edge1Dir, "edge1.properties");
+
+ private RefSpec masterRefSpec = new RefSpec("master");
+ private Optional<FSFlowTemplateCatalog> flowCatalog;
+ private Config config;
+ private BaseFlowGraph flowGraph;
+ private FsFlowGraphMonitor flowGraphMonitor;
+ private Map<URI, TopologySpec> topologySpecMap;
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ cleanUpDir(TEST_DIR.toString());
+
+
+ URI topologyCatalogUri = this.getClass().getClassLoader().getResource("topologyspec_catalog").toURI();
+ this.topologySpecMap = MultiHopFlowCompilerTest.buildTopologySpecMap(topologyCatalogUri);
+
+ this.config = ConfigBuilder.create()
+ .addPrimitive(FsFlowGraphMonitor.FS_FLOWGRAPH_MONITOR_PREFIX + "."
+ + ConfigurationKeys.FLOWGRAPH_REPO_DIR, this.flowGraphDir.getAbsolutePath())
+ .addPrimitive(FsFlowGraphMonitor.FS_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.FLOWGRAPH_BASE_DIR, "gobblin-flowgraph")
+ .addPrimitive(FsFlowGraphMonitor.FS_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, 2)
+ .build();
+
+ // Create a FSFlowTemplateCatalog instance
+ URI flowTemplateCatalogUri = this.getClass().getClassLoader().getResource("template_catalog").toURI();
+ Properties properties = new Properties();
+ this.flowGraphDir.mkdirs();
+ properties.put(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY, flowTemplateCatalogUri.toString());
+ Config config = ConfigFactory.parseProperties(properties);
+ Config templateCatalogCfg = config
+ .withValue(ConfigurationKeys.JOB_CONFIG_FILE_GENERAL_PATH_KEY,
+ config.getValue(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY));
+ this.flowCatalog = Optional.of(new FSFlowTemplateCatalog(templateCatalogCfg));
+
+ //Create a FlowGraph instance with defaults
+ this.flowGraph = new BaseFlowGraph();
+
+ this.flowGraphMonitor = new FsFlowGraphMonitor(this.config, this.flowCatalog, this.flowGraph, topologySpecMap, new CountDownLatch(1));
+ this.flowGraphMonitor.startUp();
+ this.flowGraphMonitor.setActive(true);
+ }
+
+ @Test
+ public void testAddNode() throws Exception {
+ String file1Contents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam1=value1\n";
+ String file2Contents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam2=value2\n";
+
+ addNode(this.node1Dir, this.node1File, file1Contents);
+ addNode(this.node2Dir, this.node2File, file2Contents);
+
+ // Let the monitor pick up the nodes that were recently added
+ Thread.sleep(3000);
+ for (int i = 0; i < 1; i++) {
Review Comment:
apologies if I'm being dense, but how do you verify `"param2=value2\n"`, amidst `i < 1` as the looping condition?
##########
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitorTest.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Optional;
+import com.google.common.io.Files;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import org.apache.commons.io.FileUtils;
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.DataNode;
+import org.apache.gobblin.service.modules.flowgraph.FlowEdge;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphConfigurationKeys;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.hadoop.fs.Path;
+import org.eclipse.jgit.transport.RefSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class FsFlowGraphMonitorTest {
+ private static final Logger logger = LoggerFactory.getLogger(GitFlowGraphMonitor.class);
Review Comment:
spoof! ;p
##########
gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java:
##########
@@ -1020,8 +1020,14 @@ public class ConfigurationKeys {
* Configuration properties related to flowGraphs
*/
- public static final String JAVA_PROPS_EXTENSIONS = "javaPropsExtensions";
- public static final String HOCON_FILE_EXTENSIONS = "hoconFileExtensions";
+ public static final String JAVA_PROPS_EXTENSIONS = "flowGraph.javaPropsExtensions";
+ public static final String HOCON_FILE_EXTENSIONS = "flowGraph.hoconFileExtensions";
Review Comment:
must these be mutually-exclusive? is the format comma-separated? may help to document...
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/ExceptionCatchingPathAlterationListenerDecorator.java:
##########
@@ -111,4 +111,13 @@ public void onFileDelete(Path path) {
log.error("onFileDelete failure: ", exc);
}
}
+
+ @Override
+ public void onCheckDetectedChange() {
+ try {
+ this.underlying.onCheckDetectedChange();
+ } catch (Throwable exc) {
+ log.error("onFileDelete failure: ", exc);
+ }
Review Comment:
tip: as the `on*()` methods don't throw checked exceptions, you could more succinctly implement the wrapping using a `Callable` if you define:
```
protected void logSwallowedThrowable(Callable c, String methodName) { ... }
```
note, I'm passing the name explicitly, but you could also use `Throwable.getStackTrace()` to pull out the `StackTraceElement.getMethodName()`
then call as:
```
public void onCheckDetectedChange() {
logSwallowedThrowable(() -> this.underlying.onCheckDetectedChange(), "onCheckDetectedChange");
}
```
also, note the copy-paste slip of `"onFileDelete"` above
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
+
+ private File graphDir;
+ CountDownLatch initComplete;
+
+ public FSPathAlterationFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, String baseDirectory, String flowGraphFolderName,
+ String javaPropsExtentions, String hoconFileExtensions, CountDownLatch initComplete) {
+ super(flowTemplateCatalog, graph, topologySpecMap, baseDirectory, flowGraphFolderName, javaPropsExtentions, hoconFileExtensions);
+ this.graphDir = new File(baseDirectory);
+ this.initComplete = initComplete;
+ // Populate the flowgraph with any existing files
+ if (!this.graphDir.exists()) {
+ throw new RuntimeException(String.format("Flowgraph directory at path %s does not exist!", graphDir));
+ }
+ this.populateFlowGraphAtomically();
+ }
+
+ public void onStart(final PathAlterationObserver observer) {
Review Comment:
do these need a `@Override`?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
Review Comment:
I strongly encourage class-level javadoc--usually method-level too.
e.g. I'm curious how this relates to the `FsFlowGraphMonitor` and would hope to gain insight by comparing the class descs.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
+
+ private File graphDir;
+ CountDownLatch initComplete;
+
+ public FSPathAlterationFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, String baseDirectory, String flowGraphFolderName,
+ String javaPropsExtentions, String hoconFileExtensions, CountDownLatch initComplete) {
+ super(flowTemplateCatalog, graph, topologySpecMap, baseDirectory, flowGraphFolderName, javaPropsExtentions, hoconFileExtensions);
+ this.graphDir = new File(baseDirectory);
+ this.initComplete = initComplete;
+ // Populate the flowgraph with any existing files
+ if (!this.graphDir.exists()) {
+ throw new RuntimeException(String.format("Flowgraph directory at path %s does not exist!", graphDir));
+ }
+ this.populateFlowGraphAtomically();
+ }
+
+ public void onStart(final PathAlterationObserver observer) {
+ }
+
+ public void onFileCreate(final Path path) {
+ }
+
+ public void onFileChange(final Path path) {
+ }
+
+ public void onStop(final PathAlterationObserver observer) {
+ }
+
+ public void onDirectoryCreate(final Path directory) {
+ }
+
+ public void onDirectoryChange(final Path directory) {
+ }
+
+ public void onDirectoryDelete(final Path directory) {
+ }
+
+ public void onFileDelete(final Path path) {
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ log.info("Detecting change in flowgraph files, reloading flowgraph");
+ this.populateFlowGraphAtomically();
+ }
+
+ private void populateFlowGraphAtomically() {
+ FlowGraph newFlowGraph = new BaseFlowGraph();
+ try {
+ List<Path> edges = new ArrayList<>();
+ // All nodes must be added first before edges, otherwise edges may have a missing source or destination.
+ Files.walk(this.graphDir.toPath()).forEach(fileName -> {
+ if (!Files.isDirectory(fileName)) {
+ if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), NODE_FILE_DEPTH)) {
+ addDataNode(newFlowGraph, fileName.toString());
+ } else if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), EDGE_FILE_DEPTH)) {
+ edges.add(new Path(fileName.toString()));
+ }
+ }
+ });
+ for (Path edge: edges) {
+ addFlowEdge(newFlowGraph, edge.toString());
+ }
+ // Reduce the countdown latch
+ this.initComplete.countDown();
+ this.flowGraph.copyGraph(newFlowGraph);
Review Comment:
the pattern seems to be: prepare another flow graph then use its state to reinitialize the active flow graph. why not instead just make the new flow graph itself active? basically an atomic reference swap. that approach would allow flow graphs to be immutable, which is generally a Good Thing: there's no need for locking and any in-flight reads continue against the one they started with, even if a different FG becomes active in the meanwhile.
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/PathAlterationObserver.java:
##########
@@ -43,9 +43,9 @@ public class PathAlterationObserver {
private final PathFilter pathFilter;
private final Comparator<Path> comparator;
private final FileSystem fs;
-
private final Path[] EMPTY_PATH_ARRAY = new Path[0];
+ private boolean changeApplied = false;
Review Comment:
since only `checkAndNotify` is `synchronized`, not the others setting this, it should be `volatile`.
that said, the "check-then-act" of `if (this.changeApplied) {...}` may be unsafe until those other methods actually are `synchronized`.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private boolean isActive;
+ private long pollingInterval;
+ private PathAlterationObserverScheduler pathAlterationDetector;
+ private PathAlterationObserver observer;
+ private Path flowGraphPath;
+
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS)
+ .build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete) throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval = TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
Review Comment:
not 100% sure here... may deserve a comment... is this some legacy compatability dual mode of supporting the git repo in addition, so either may be used?
if so, consider a config to disable that once we're ready to move entirely toward the FS dir
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r975690002
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -197,10 +200,9 @@ private boolean checkFilePath(String file, int depth) {
Path filePath = new Path(file);
Review Comment:
It's a relative path to the root of the flowgraph, I'll update the javadoc to reflect this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] AndyJiang99 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
AndyJiang99 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r975881365
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/ExceptionCatchingPathAlterationListenerDecorator.java:
##########
@@ -42,73 +45,83 @@ public Object getDecoratedObject() {
@Override
public void onStart(PathAlterationObserver observer) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onStart(observer);
- } catch (Throwable exc) {
- log.error("onStart failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileCreate(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileCreate(path);
- } catch (Throwable exc) {
- log.error("onFileCreate failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileChange(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileChange(path);
- } catch (Throwable exc) {
- log.error("onFileChange failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onStop(PathAlterationObserver observer) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onStop(observer);
- } catch (Throwable exc) {
- log.error("onStop failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryCreate(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryCreate(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryCreate failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryChange(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryChange(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryChange failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onDirectoryDelete(Path directory) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onDirectoryDelete(directory);
- } catch (Throwable exc) {
- log.error("onDirectoryDelete failure: ", exc);
- }
+ return null;
+ });
}
@Override
public void onFileDelete(Path path) {
- try {
+ logSwallowedThrowable(() -> {
this.underlying.onFileDelete(path);
+ return null;
+ });
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ logSwallowedThrowable(() -> {
+ this.underlying.onCheckDetectedChange();
+ return null;
+ });
+ }
+
+ protected void logSwallowedThrowable(Callable<Void> c) {
Review Comment:
Ideally a better argument variable name than just `c`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r977047298
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraphHelper;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static final long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private volatile boolean isActive = false;
+ private final long pollingInterval;
+ private BaseFlowGraphHelper flowGraphHelper;
+ private final PathAlterationObserverScheduler pathAlterationDetector;
+ private final FSPathAlterationFlowGraphListener listener;
+ private final PathAlterationObserver observer;
+ private final Path flowGraphPath;
+ private final AtomicReference<FlowGraph> flowGraph;
+ private final CountDownLatch initComplete;
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS).build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ AtomicReference<FlowGraph> graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete)
+ throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval =
+ TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
+ this.flowGraphHelper = new BaseFlowGraphHelper(flowTemplateCatalog, topologySpecMap, flowGraphPath.toString(),
+ configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_BASE_DIR), configWithFallbacks.getString(ConfigurationKeys.JAVA_PROPS_EXTENSIONS),
+ configWithFallbacks.getString(ConfigurationKeys.HOCON_FILE_EXTENSIONS));
+ this.listener = new FSPathAlterationFlowGraphListener(flowTemplateCatalog, graph, flowGraphPath.toString(), this.flowGraphHelper);
+
+ this.flowGraph = graph;
+ this.initComplete = initComplete;
+
+ if (pollingInterval == ConfigurationKeys.DISABLED_JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL) {
+ this.pathAlterationDetector = null;
+ } else {
+ this.pathAlterationDetector = new PathAlterationObserverScheduler(pollingInterval);
+ Optional<PathAlterationObserver> observerOptional = Optional.fromNullable(observer);
+ this.pathAlterationDetector.addPathAlterationObserver(this.listener, observerOptional,
+ this.flowGraphPath);
+ }
+ }
+
+ @Override
+ protected void startUp()
+ throws IOException {
+ }
+
+ @Override
+ public synchronized void setActive(boolean isActive) {
+ log.info("Setting the flow graph monitor to be " + isActive + " from " + this.isActive);
+ if (this.isActive == isActive) {
+ // No-op if already in correct state
+ return;
+ } else if (isActive) {
+ if (this.pathAlterationDetector != null) {
+ log.info("Starting the " + getClass().getSimpleName());
+ log.info("Polling flowgraph folder with interval {} ", this.pollingInterval);
+ try {
+ this.pathAlterationDetector.start();
+ // Manually instantiate flowgraph when the monitor becomes active
+ this.flowGraphHelper.populateFlowGraphAtomically(this.flowGraph);
+ // Reduce the countdown latch
+ this.initComplete.countDown();
+ log.info("Finished populating FSFlowgraph");
+ } catch (IOException e) {
+ log.error("Could not initialize pathAlterationDetector due to error: ", e);
+ }
+ } else {
+ log.warn("No path alteration detector found");
+ }
+ }
+ this.isActive = isActive;
Review Comment:
Need to consider the scenario where both isActive == false and this.isActive == true, if it gets moved to the else statement it only considers isActive == true and this.isActive = false
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970090280
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
+
+ private File graphDir;
+ CountDownLatch initComplete;
+
+ public FSPathAlterationFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, String baseDirectory, String flowGraphFolderName,
+ String javaPropsExtentions, String hoconFileExtensions, CountDownLatch initComplete) {
+ super(flowTemplateCatalog, graph, topologySpecMap, baseDirectory, flowGraphFolderName, javaPropsExtentions, hoconFileExtensions);
+ this.graphDir = new File(baseDirectory);
+ this.initComplete = initComplete;
+ // Populate the flowgraph with any existing files
+ if (!this.graphDir.exists()) {
+ throw new RuntimeException(String.format("Flowgraph directory at path %s does not exist!", graphDir));
+ }
+ this.populateFlowGraphAtomically();
+ }
+
+ public void onStart(final PathAlterationObserver observer) {
+ }
+
+ public void onFileCreate(final Path path) {
+ }
+
+ public void onFileChange(final Path path) {
+ }
+
+ public void onStop(final PathAlterationObserver observer) {
+ }
+
+ public void onDirectoryCreate(final Path directory) {
+ }
+
+ public void onDirectoryChange(final Path directory) {
+ }
+
+ public void onDirectoryDelete(final Path directory) {
+ }
+
+ public void onFileDelete(final Path path) {
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ log.info("Detecting change in flowgraph files, reloading flowgraph");
+ this.populateFlowGraphAtomically();
+ }
+
+ private void populateFlowGraphAtomically() {
+ FlowGraph newFlowGraph = new BaseFlowGraph();
+ try {
+ List<Path> edges = new ArrayList<>();
+ // All nodes must be added first before edges, otherwise edges may have a missing source or destination.
+ Files.walk(this.graphDir.toPath()).forEach(fileName -> {
+ if (!Files.isDirectory(fileName)) {
+ if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), NODE_FILE_DEPTH)) {
+ addDataNode(newFlowGraph, fileName.toString());
+ } else if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), EDGE_FILE_DEPTH)) {
+ edges.add(new Path(fileName.toString()));
+ }
+ }
+ });
+ for (Path edge: edges) {
+ addFlowEdge(newFlowGraph, edge.toString());
Review Comment:
Unfortunately it's due to the implementation of the file listener, it was originally written for Gobblin pipelines and thus it uses the Hadoop Path references instead of Java's nio.file.path Path class. There are a lot of usages of Hadoop Path objects in GaaS that I find unnecessary, especially because doing so requires GaaS to stay on outdated versions of Java. Hence I try to avoid having new classes rely on the Hadoop Path library but for compatibility it seems hard to avoid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970178249
##########
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitorTest.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Optional;
+import com.google.common.io.Files;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import org.apache.commons.io.FileUtils;
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.DataNode;
+import org.apache.gobblin.service.modules.flowgraph.FlowEdge;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphConfigurationKeys;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.hadoop.fs.Path;
+import org.eclipse.jgit.transport.RefSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class FsFlowGraphMonitorTest {
+ private static final Logger logger = LoggerFactory.getLogger(GitFlowGraphMonitor.class);
+ private final File TEST_DIR = new File(FileUtils.getTempDirectory(), "fsFlowGraphTestDir");
+ private final File flowGraphDir = new File(TEST_DIR, "/gobblin-flowgraph");
+ private static final String NODE_1_FILE = "node1.properties";
+ private final File node1Dir = new File(FileUtils.getTempDirectory(), "node1");
+ private final File node1File = new File(node1Dir, NODE_1_FILE);
+ private static final String NODE_2_FILE = "node2.properties";
+ private final File node2Dir = new File(FileUtils.getTempDirectory(), "node2");
+ private final File node2File = new File(node2Dir, NODE_2_FILE);
+ private final File edge1Dir = new File(node1Dir, "node2");
+ private final File edge1File = new File(edge1Dir, "edge1.properties");
+
+ private RefSpec masterRefSpec = new RefSpec("master");
+ private Optional<FSFlowTemplateCatalog> flowCatalog;
+ private Config config;
+ private BaseFlowGraph flowGraph;
+ private FsFlowGraphMonitor flowGraphMonitor;
+ private Map<URI, TopologySpec> topologySpecMap;
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ cleanUpDir(TEST_DIR.toString());
+
+
+ URI topologyCatalogUri = this.getClass().getClassLoader().getResource("topologyspec_catalog").toURI();
+ this.topologySpecMap = MultiHopFlowCompilerTest.buildTopologySpecMap(topologyCatalogUri);
+
+ this.config = ConfigBuilder.create()
+ .addPrimitive(FsFlowGraphMonitor.FS_FLOWGRAPH_MONITOR_PREFIX + "."
+ + ConfigurationKeys.FLOWGRAPH_REPO_DIR, this.flowGraphDir.getAbsolutePath())
+ .addPrimitive(FsFlowGraphMonitor.FS_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.FLOWGRAPH_BASE_DIR, "gobblin-flowgraph")
+ .addPrimitive(FsFlowGraphMonitor.FS_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, 2)
+ .build();
+
+ // Create a FSFlowTemplateCatalog instance
+ URI flowTemplateCatalogUri = this.getClass().getClassLoader().getResource("template_catalog").toURI();
+ Properties properties = new Properties();
+ this.flowGraphDir.mkdirs();
+ properties.put(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY, flowTemplateCatalogUri.toString());
+ Config config = ConfigFactory.parseProperties(properties);
+ Config templateCatalogCfg = config
+ .withValue(ConfigurationKeys.JOB_CONFIG_FILE_GENERAL_PATH_KEY,
+ config.getValue(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY));
+ this.flowCatalog = Optional.of(new FSFlowTemplateCatalog(templateCatalogCfg));
+
+ //Create a FlowGraph instance with defaults
+ this.flowGraph = new BaseFlowGraph();
+
+ this.flowGraphMonitor = new FsFlowGraphMonitor(this.config, this.flowCatalog, this.flowGraph, topologySpecMap, new CountDownLatch(1));
+ this.flowGraphMonitor.startUp();
+ this.flowGraphMonitor.setActive(true);
+ }
+
+ @Test
+ public void testAddNode() throws Exception {
+ String file1Contents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam1=value1\n";
+ String file2Contents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam2=value2\n";
+
+ addNode(this.node1Dir, this.node1File, file1Contents);
+ addNode(this.node2Dir, this.node2File, file2Contents);
+
+ // Let the monitor pick up the nodes that were recently added
+ Thread.sleep(3000);
+ for (int i = 0; i < 1; i++) {
Review Comment:
Ah shoot I think I set the loop differently to test something else but forgot to fix, thanks for the callout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970168700
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/PathAlterationObserver.java:
##########
@@ -43,9 +43,9 @@ public class PathAlterationObserver {
private final PathFilter pathFilter;
private final Comparator<Path> comparator;
private final FileSystem fs;
-
private final Path[] EMPTY_PATH_ARRAY = new Path[0];
+ private boolean changeApplied = false;
Review Comment:
I'll change the other functions to be synchronized as well, but I was under the impression it would be safe since those other functions are dependent on checkAndNotify, and since only 1 function execution can run at once, then the others shouldn't be able to modify the check at the same time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970155176
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/ExceptionCatchingPathAlterationListenerDecorator.java:
##########
@@ -111,4 +111,13 @@ public void onFileDelete(Path path) {
log.error("onFileDelete failure: ", exc);
}
}
+
+ @Override
+ public void onCheckDetectedChange() {
+ try {
+ this.underlying.onCheckDetectedChange();
+ } catch (Throwable exc) {
+ log.error("onFileDelete failure: ", exc);
+ }
Review Comment:
Oops good catch, will try to implement this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r977046165
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraphHelper;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static final long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private volatile boolean isActive = false;
+ private final long pollingInterval;
+ private BaseFlowGraphHelper flowGraphHelper;
+ private final PathAlterationObserverScheduler pathAlterationDetector;
+ private final FSPathAlterationFlowGraphListener listener;
+ private final PathAlterationObserver observer;
+ private final Path flowGraphPath;
+ private final AtomicReference<FlowGraph> flowGraph;
+ private final CountDownLatch initComplete;
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS).build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ AtomicReference<FlowGraph> graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete)
+ throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval =
+ TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
+ this.flowGraphHelper = new BaseFlowGraphHelper(flowTemplateCatalog, topologySpecMap, flowGraphPath.toString(),
+ configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_BASE_DIR), configWithFallbacks.getString(ConfigurationKeys.JAVA_PROPS_EXTENSIONS),
+ configWithFallbacks.getString(ConfigurationKeys.HOCON_FILE_EXTENSIONS));
+ this.listener = new FSPathAlterationFlowGraphListener(flowTemplateCatalog, graph, flowGraphPath.toString(), this.flowGraphHelper);
+
+ this.flowGraph = graph;
+ this.initComplete = initComplete;
+
+ if (pollingInterval == ConfigurationKeys.DISABLED_JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL) {
+ this.pathAlterationDetector = null;
+ } else {
+ this.pathAlterationDetector = new PathAlterationObserverScheduler(pollingInterval);
+ Optional<PathAlterationObserver> observerOptional = Optional.fromNullable(observer);
+ this.pathAlterationDetector.addPathAlterationObserver(this.listener, observerOptional,
+ this.flowGraphPath);
+ }
+ }
+
+ @Override
+ protected void startUp()
+ throws IOException {
+ }
+
+ @Override
+ public synchronized void setActive(boolean isActive) {
+ log.info("Setting the flow graph monitor to be " + isActive + " from " + this.isActive);
+ if (this.isActive == isActive) {
+ // No-op if already in correct state
+ return;
+ } else if (isActive) {
+ if (this.pathAlterationDetector != null) {
+ log.info("Starting the " + getClass().getSimpleName());
+ log.info("Polling flowgraph folder with interval {} ", this.pollingInterval);
+ try {
+ this.pathAlterationDetector.start();
+ // Manually instantiate flowgraph when the monitor becomes active
+ this.flowGraphHelper.populateFlowGraphAtomically(this.flowGraph);
+ // Reduce the countdown latch
+ this.initComplete.countDown();
+ log.info("Finished populating FSFlowgraph");
+ } catch (IOException e) {
+ log.error("Could not initialize pathAlterationDetector due to error: ", e);
+ }
+ } else {
+ log.warn("No path alteration detector found");
+ }
+ }
+ this.isActive = isActive;
Review Comment:
It's representing a generic callable so I think a short name is fine here, usually we want more descriptive names for contextual variables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] phet commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
phet commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r977194076
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flow/MultiHopFlowCompiler.java:
##########
@@ -82,7 +85,7 @@
@Slf4j
public class MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler {
@Getter
- private final FlowGraph flowGraph;
+ private final AtomicReference<FlowGraph> flowGraph;
Review Comment:
`@Getter` may break encapsulation more than desired: let's not invite callers to adjust our reference. better perhaps to define an accessor returning `AtomicReference.get` to them.
in order to handle atomic updates, define a setter (yourself, `@Setter` won't quite work)
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flow/MultiHopFlowCompiler.java:
##########
@@ -168,7 +171,7 @@ public MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrum
}
@VisibleForTesting
- MultiHopFlowCompiler(Config config, FlowGraph flowGraph) {
+ MultiHopFlowCompiler(Config config, AtomicReference<FlowGraph> flowGraph) {
Review Comment:
similar to comments above around encapsulation, more expected would be a `FlowGraph` param for the MHFC to store internally into the `AtomicReference`.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -277,32 +282,64 @@ private List<SpecExecutor> getSpecExecutors(Config edgeConfig) throws URISyntaxE
* @return the configuration object
* @throws IOException
*/
- protected Config loadNodeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadNodeFileWithOverrides(Path filePath)
+ throws IOException {
Config nodeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getNodeConfigWithOverrides(nodeConfig, filePath);
}
-
/**
* Load the edge file.
* @param filePath path of the edge file relative to the repository root
* @return the configuration object
* @throws IOException
*/
- protected Config loadEdgeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadEdgeFileWithOverrides(Path filePath)
+ throws IOException {
Config edgeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getEdgeConfigWithOverrides(edgeConfig, filePath);
}
+ /**
+ * Loads the entire flowgraph from the path configured in {@link org.apache.gobblin.configuration.ConfigurationKeys.FLOWGRAPH_BASE_DIR }
+ * Expects nodes to be in the format of /flowGraphName/nodeA/nodeA.properties
+ * Expects edges to be in the format of /flowGraphName/nodeA/nodeB/edgeAB.properties
+ * The current flowgraph will be swapped atomically with the new flowgraph that is loaded
+ */
+ public void populateFlowGraphAtomically(AtomicReference<FlowGraph> flowGraphReference) {
+ FlowGraph newFlowGraph = new BaseFlowGraph();
+ java.nio.file.Path graphPath = new File(this.baseDirectory).toPath();
+ try {
+ List<Path> edges = new ArrayList<>();
+ // All nodes must be added first before edges, otherwise edges may have a missing source or destination.
+ // Need to convert files to Hadoop Paths to be compatible with FileAlterationListener
+ java.nio.file.Files.walk(graphPath).forEach(fileName -> {
+ if (!java.nio.file.Files.isDirectory(fileName)) {
+ if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), NODE_FILE_DEPTH)) {
+ addDataNode(newFlowGraph, fileName.toString());
+ } else if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), EDGE_FILE_DEPTH)) {
+ edges.add(new Path(fileName.toString()));
+ }
Review Comment:
I wonder what might be silently skipped here... whether logging might assist in debugging violated presumptions
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -277,32 +282,64 @@ private List<SpecExecutor> getSpecExecutors(Config edgeConfig) throws URISyntaxE
* @return the configuration object
* @throws IOException
*/
- protected Config loadNodeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadNodeFileWithOverrides(Path filePath)
+ throws IOException {
Config nodeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getNodeConfigWithOverrides(nodeConfig, filePath);
}
-
/**
* Load the edge file.
* @param filePath path of the edge file relative to the repository root
* @return the configuration object
* @throws IOException
*/
- protected Config loadEdgeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadEdgeFileWithOverrides(Path filePath)
+ throws IOException {
Config edgeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getEdgeConfigWithOverrides(edgeConfig, filePath);
}
+ /**
+ * Loads the entire flowgraph from the path configured in {@link org.apache.gobblin.configuration.ConfigurationKeys.FLOWGRAPH_BASE_DIR }
+ * Expects nodes to be in the format of /flowGraphName/nodeA/nodeA.properties
+ * Expects edges to be in the format of /flowGraphName/nodeA/nodeB/edgeAB.properties
+ * The current flowgraph will be swapped atomically with the new flowgraph that is loaded
+ */
+ public void populateFlowGraphAtomically(AtomicReference<FlowGraph> flowGraphReference) {
Review Comment:
rather than "handing over the keys", and combining responsibility for loading w/ managing orderly updates, why not merely return the loaded `FlowGraph` and leave it to the caller to handle update/replacement?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -84,26 +84,27 @@ public BaseFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempl
this.hoconFileExtensions = Sets.newHashSet(hoconFileExtensions.split(","));
try {
this.pullFileLoader = new PullFileLoader(folderPath,
- FileSystem.get(URI.create(ConfigurationKeys.LOCAL_FS_URI), new Configuration()),
- this.javaPropsExtensions, this.hoconFileExtensions);
+ FileSystem.get(URI.create(ConfigurationKeys.LOCAL_FS_URI), new Configuration()), this.javaPropsExtensions,
+ this.hoconFileExtensions);
} catch (IOException e) {
throw new RuntimeException("Could not create pull file loader", e);
}
}
+
/**
* Add a {@link DataNode} to the {@link FlowGraph}. The method uses the {@link FlowGraphConfigurationKeys#DATA_NODE_CLASS} config
* to instantiate a {@link DataNode} from the node config file.
* @param path of node to add
*/
- protected void addDataNode(String path) {
+ public void addDataNode(FlowGraph graph, String path) {
if (checkFilePath(path, NODE_FILE_DEPTH)) {
Path nodeFilePath = new Path(this.baseDirectory, path);
try {
Config config = loadNodeFileWithOverrides(nodeFilePath);
Class dataNodeClass = Class.forName(ConfigUtils.getString(config, FlowGraphConfigurationKeys.DATA_NODE_CLASS,
FlowGraphConfigurationKeys.DEFAULT_DATA_NODE_CLASS));
DataNode dataNode = (DataNode) GobblinConstructorUtils.invokeLongestConstructor(dataNodeClass, config);
- if (!this.flowGraph.addDataNode(dataNode)) {
+ if (!graph.addDataNode(dataNode)) {
Review Comment:
overall I don't want to slow down the refactoring, so up to you on whether or how to address...
this helper's methods all follow the same pattern, which appears a clunky abstraction: they convert a `path` into whatever argument for a call on `flowGraph` that they delegate to the same. it's as if we wished the `FlowGraph` methods to take a `String path` as their lone arg--and have that magically work
Since the gap boils down merely to conversion (and the `FlowGraph` itself is not involved there), I'd propose a thinner, more cohesive abstraction: a few converter/factory methods, each taking a `path` and returning a `DataNode` or a `FlowEdge` or a `String flowId`.
impl-wise, all would leverage one core method to resolve `String path` to `Config`. the latter motivates the instance state of the pull loader and the base dir. that's a "path-based config resolver".
separately, if we wish to support the pattern of:
```
if (!flowGraph.op(arg)) {
log("op problem w/ {}", arg.id);
} else {
log("it worked w/ {}", arg.id)
}
```
write a logging `FlowGraph` Decorator that wraps each such method.
edit: see further comments below...
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraph.java:
##########
@@ -47,10 +47,10 @@
public class BaseFlowGraph implements FlowGraph {
private final ReadWriteLock rwLock = new ReentrantReadWriteLock(true);
Review Comment:
while you're documenting the rationale for `AtomicReference` (suggested by zihan), let's also clarify what particular instance state this lock means to protect.
...at this point, I don't see non-`final` fields in the object... although I do note the maps are not immutable.
that last part is significant: even with careful atomic access and update via the reference, the FG can mutate out from under whomever is using it!
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraphHelper;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static final long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private volatile boolean isActive = false;
+ private final long pollingInterval;
+ private BaseFlowGraphHelper flowGraphHelper;
+ private final PathAlterationObserverScheduler pathAlterationDetector;
+ private final FSPathAlterationFlowGraphListener listener;
+ private final PathAlterationObserver observer;
+ private final Path flowGraphPath;
+ private final AtomicReference<FlowGraph> flowGraph;
+ private final CountDownLatch initComplete;
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS).build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ AtomicReference<FlowGraph> graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete)
+ throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval =
+ TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
+ this.flowGraphHelper = new BaseFlowGraphHelper(flowTemplateCatalog, topologySpecMap, flowGraphPath.toString(),
+ configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_BASE_DIR), configWithFallbacks.getString(ConfigurationKeys.JAVA_PROPS_EXTENSIONS),
+ configWithFallbacks.getString(ConfigurationKeys.HOCON_FILE_EXTENSIONS));
+ this.listener = new FSPathAlterationFlowGraphListener(flowTemplateCatalog, graph, flowGraphPath.toString(), this.flowGraphHelper);
+
+ this.flowGraph = graph;
+ this.initComplete = initComplete;
+
+ if (pollingInterval == ConfigurationKeys.DISABLED_JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL) {
+ this.pathAlterationDetector = null;
+ } else {
+ this.pathAlterationDetector = new PathAlterationObserverScheduler(pollingInterval);
+ Optional<PathAlterationObserver> observerOptional = Optional.fromNullable(observer);
+ this.pathAlterationDetector.addPathAlterationObserver(this.listener, observerOptional,
+ this.flowGraphPath);
+ }
+ }
+
+ @Override
+ protected void startUp()
+ throws IOException {
+ }
+
+ @Override
+ public synchronized void setActive(boolean isActive) {
+ log.info("Setting the flow graph monitor to be " + isActive + " from " + this.isActive);
+ if (this.isActive == isActive) {
+ // No-op if already in correct state
+ return;
+ } else if (isActive) {
+ if (this.pathAlterationDetector != null) {
+ log.info("Starting the " + getClass().getSimpleName());
+ log.info("Polling flowgraph folder with interval {} ", this.pollingInterval);
+ try {
+ this.pathAlterationDetector.start();
+ // Manually instantiate flowgraph when the monitor becomes active
+ this.flowGraphHelper.populateFlowGraphAtomically(this.flowGraph);
Review Comment:
once I read these FlowGraphMonitors, I realized why the `BaseFlowGraphHelper.addDataNode()` and other methods are `public`. in essence, the `flowGraphHelper` unequally serves this FS monitor over the Git one, by providing this method only for the former, with no equivalent for the latter: instead similar discernment of path nesting depth on the Git side is within `GitFlowGraphListener`.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/GitFlowGraphListener.java:
##########
@@ -53,10 +47,10 @@ public GitFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempla
@Override
public void addChange(DiffEntry change) {
Path path = new Path(change.getNewPath());
- if (path.depth() == NODE_FILE_DEPTH) {
- addDataNode(change.getNewPath());
- } else if (path.depth() == EDGE_FILE_DEPTH) {
- addFlowEdge(change.getNewPath());
+ if (path.depth() == BaseFlowGraphHelper.NODE_FILE_DEPTH) {
+ this.baseFlowGraphHelper.addDataNode(this.flowGraph.get(), change.getNewPath());
Review Comment:
the code here illustrates the challenge: even though the `FlowGraph` is mediated by an `AtomicReference`, it remains mutable. one irresponsible actor could bring ruin to everyone!
ideally instead of updating the *same* flow graph, we'd rather create a updated *copy* of it, reflecting the differences.
both because that has cost and because we'd like to enjoy the atomic semantics we're working so hard for, we wouldn't do this on each addition/removal, but rather one time per invocation of `GitFlowGraphMonitor.processGitConfigChanges()`.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.modules.flowgraph;
+
+import java.io.File;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+
+
+/**
+ * An implementation of {@link PathAlterationListener} to listen for changes in a directory and apply it to a GaaS FlowGraph
+ * Is invoked by {@link PathAlterationObserver} which would check a folder and perform recursive comparisons on files compared to
+ * their last polled state. On any detected differences in files when a check is done, the {@link FlowGraph} will be updated.
+ *
+ * Unlike the {@link GitFlowGraphListener}, this class will reload the entire flowgraph on any detected change, instead loading only the diffs.
+ */
+@Slf4j
+public class FSPathAlterationFlowGraphListener implements PathAlterationListener {
+ private final AtomicReference<FlowGraph> flowGraph;
Review Comment:
I'm surprised to see another `AtomicReference` over here. is the same one from the MHFC... basically entrusted over to this class? I recommend instead that this listener hold a reference to the MHFC itself and invoke a setter method for encapsulating the `AtomicReference`.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitor.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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 java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.flowgraph.FSPathAlterationFlowGraphListener;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphMonitor;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.gobblin.util.filesystem.PathAlterationObserverScheduler;
+
+
+@Slf4j
+public class FsFlowGraphMonitor extends AbstractIdleService implements FlowGraphMonitor {
+ public static final String FS_FLOWGRAPH_MONITOR_PREFIX = "gobblin.service.fsFlowGraphMonitor";
+ private static long DEFAULT_FLOWGRAPH_POLLING_INTERVAL = 60;
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR = "git-flowgraph";
+ private static final String DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR = "gobblin-flowgraph";
+ private boolean isActive;
+ private long pollingInterval;
+ private PathAlterationObserverScheduler pathAlterationDetector;
+ private PathAlterationObserver observer;
+ private Path flowGraphPath;
+
+ private static final Config DEFAULT_FALLBACK = ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
+ .put(ConfigurationKeys.FLOWGRAPH_REPO_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_REPO_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_BASE_DIR, DEFAULT_FS_FLOWGRAPH_MONITOR_FLOWGRAPH_DIR)
+ .put(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL, DEFAULT_FLOWGRAPH_POLLING_INTERVAL)
+ .put(ConfigurationKeys.JAVA_PROPS_EXTENSIONS, ConfigurationKeys.DEFAULT_PROPERTIES_EXTENSIONS)
+ .put(ConfigurationKeys.HOCON_FILE_EXTENSIONS, ConfigurationKeys.DEFAULT_CONF_EXTENSIONS)
+ .build());
+
+ public FsFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete) throws IOException {
+ Config configWithFallbacks = config.getConfig(FS_FLOWGRAPH_MONITOR_PREFIX).withFallback(DEFAULT_FALLBACK);
+ this.pollingInterval = TimeUnit.SECONDS.toMillis(configWithFallbacks.getLong(ConfigurationKeys.FLOWGRAPH_POLLING_INTERVAL));
+ this.flowGraphPath = new Path(configWithFallbacks.getString(ConfigurationKeys.FLOWGRAPH_REPO_DIR));
+ this.observer = new PathAlterationObserver(flowGraphPath);
Review Comment:
yes, probably better to create config keys specific to this
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.modules.flowgraph;
+
+import java.io.File;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Optional;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+
+
+/**
+ * An implementation of {@link PathAlterationListener} to listen for changes in a directory and apply it to a GaaS FlowGraph
+ * Is invoked by {@link PathAlterationObserver} which would check a folder and perform recursive comparisons on files compared to
+ * their last polled state. On any detected differences in files when a check is done, the {@link FlowGraph} will be updated.
+ *
+ * Unlike the {@link GitFlowGraphListener}, this class will reload the entire flowgraph on any detected change, instead loading only the diffs.
Review Comment:
you may mean "instead *of* loading only the diffs"
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/PathAlterationObserver.java:
##########
@@ -43,9 +43,9 @@ public class PathAlterationObserver {
private final PathFilter pathFilter;
private final Comparator<Path> comparator;
private final FileSystem fs;
-
private final Path[] EMPTY_PATH_ARRAY = new Path[0];
+ private boolean changeApplied = false;
Review Comment:
that's true: if all the others are called only within `checkThenNotify`, which is `synchronized`, then it's not necessary to synchronize them in addition. I hadn't grasped earlier whether that's the case.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -277,32 +282,64 @@ private List<SpecExecutor> getSpecExecutors(Config edgeConfig) throws URISyntaxE
* @return the configuration object
* @throws IOException
*/
- protected Config loadNodeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadNodeFileWithOverrides(Path filePath)
+ throws IOException {
Config nodeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getNodeConfigWithOverrides(nodeConfig, filePath);
}
-
/**
* Load the edge file.
* @param filePath path of the edge file relative to the repository root
* @return the configuration object
* @throws IOException
*/
- protected Config loadEdgeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadEdgeFileWithOverrides(Path filePath)
+ throws IOException {
Config edgeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getEdgeConfigWithOverrides(edgeConfig, filePath);
}
+ /**
+ * Loads the entire flowgraph from the path configured in {@link org.apache.gobblin.configuration.ConfigurationKeys.FLOWGRAPH_BASE_DIR }
+ * Expects nodes to be in the format of /flowGraphName/nodeA/nodeA.properties
+ * Expects edges to be in the format of /flowGraphName/nodeA/nodeB/edgeAB.properties
+ * The current flowgraph will be swapped atomically with the new flowgraph that is loaded
+ */
+ public void populateFlowGraphAtomically(AtomicReference<FlowGraph> flowGraphReference) {
Review Comment:
also, I suggested earlier to retool the abstraction to avoid wrapping the `FlowGraph` method invocations. seeing this method, I finally appreciate that it is the core abstraction--basically you've got a flow graph initializer/loader/factory. I was thrown off earlier by the other methods being `public`, when they really seem subordinate to this one.
if you made them protected or private, renamed this class, and placed this method first, I don't find much need to rework the abstraction.
that said, this method seems very tied to the FS impl, and not relevant to the Git one. if so, that might be an argument for the thinner abstraction of a "path-based config loader" shared between both. this functionality here would be FS-specific, so not within the same class.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/GitFlowGraphListener.java:
##########
@@ -53,10 +47,10 @@ public GitFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempla
@Override
public void addChange(DiffEntry change) {
Path path = new Path(change.getNewPath());
- if (path.depth() == NODE_FILE_DEPTH) {
- addDataNode(change.getNewPath());
- } else if (path.depth() == EDGE_FILE_DEPTH) {
- addFlowEdge(change.getNewPath());
+ if (path.depth() == BaseFlowGraphHelper.NODE_FILE_DEPTH) {
+ this.baseFlowGraphHelper.addDataNode(this.flowGraph.get(), change.getNewPath());
Review Comment:
the difficulty here is that the java collections lib has such weak support for immutable collections, specifically for sharing common parts between instances (consider instead the purely functional data structures of scala or clojure). what this means is that the copy-on-write is more expensive... still, the approach nonetheless works, just "higher tax".
really we have two choices:
a. keep the flow graph itself immutable and allow it to be cloned into a builder. once all adjustments are made, the builder's `.build()` renders a new immutable instance.
b. have a scratch flow graph not directly used by the MHFC, that is mutable. its 'add' and 'remove' methods would be `synchronzied`. additionally have a `synchronized` `createSnapshot()` method that creates a copy of the current state, which is an immutable flow graph.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r978337541
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/GitFlowGraphListener.java:
##########
@@ -53,10 +47,10 @@ public GitFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempla
@Override
public void addChange(DiffEntry change) {
Path path = new Path(change.getNewPath());
- if (path.depth() == NODE_FILE_DEPTH) {
- addDataNode(change.getNewPath());
- } else if (path.depth() == EDGE_FILE_DEPTH) {
- addFlowEdge(change.getNewPath());
+ if (path.depth() == BaseFlowGraphHelper.NODE_FILE_DEPTH) {
+ this.baseFlowGraphHelper.addDataNode(this.flowGraph.get(), change.getNewPath());
Review Comment:
I'm going to lean the latter approach which seems simpler to implement. Reason being is that I don't think that it is possible for the Git Flowgraph to run into issues with concurrent updates given the use of it (though it is theoretical in practice) and also due to that this entire module will be deprecated soon anyways due to the FSFlowgraph.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970090280
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/FSPathAlterationFlowGraphListener.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.modules.flowgraph;
+
+import com.google.common.base.Optional;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.gobblin.util.filesystem.PathAlterationListener;
+import org.apache.gobblin.util.filesystem.PathAlterationObserver;
+import org.apache.hadoop.fs.Path;
+
+@Slf4j
+public class FSPathAlterationFlowGraphListener extends BaseFlowGraphListener implements PathAlterationListener {
+
+ private File graphDir;
+ CountDownLatch initComplete;
+
+ public FSPathAlterationFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
+ FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, String baseDirectory, String flowGraphFolderName,
+ String javaPropsExtentions, String hoconFileExtensions, CountDownLatch initComplete) {
+ super(flowTemplateCatalog, graph, topologySpecMap, baseDirectory, flowGraphFolderName, javaPropsExtentions, hoconFileExtensions);
+ this.graphDir = new File(baseDirectory);
+ this.initComplete = initComplete;
+ // Populate the flowgraph with any existing files
+ if (!this.graphDir.exists()) {
+ throw new RuntimeException(String.format("Flowgraph directory at path %s does not exist!", graphDir));
+ }
+ this.populateFlowGraphAtomically();
+ }
+
+ public void onStart(final PathAlterationObserver observer) {
+ }
+
+ public void onFileCreate(final Path path) {
+ }
+
+ public void onFileChange(final Path path) {
+ }
+
+ public void onStop(final PathAlterationObserver observer) {
+ }
+
+ public void onDirectoryCreate(final Path directory) {
+ }
+
+ public void onDirectoryChange(final Path directory) {
+ }
+
+ public void onDirectoryDelete(final Path directory) {
+ }
+
+ public void onFileDelete(final Path path) {
+ }
+
+ @Override
+ public void onCheckDetectedChange() {
+ log.info("Detecting change in flowgraph files, reloading flowgraph");
+ this.populateFlowGraphAtomically();
+ }
+
+ private void populateFlowGraphAtomically() {
+ FlowGraph newFlowGraph = new BaseFlowGraph();
+ try {
+ List<Path> edges = new ArrayList<>();
+ // All nodes must be added first before edges, otherwise edges may have a missing source or destination.
+ Files.walk(this.graphDir.toPath()).forEach(fileName -> {
+ if (!Files.isDirectory(fileName)) {
+ if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), NODE_FILE_DEPTH)) {
+ addDataNode(newFlowGraph, fileName.toString());
+ } else if (checkFileLevelRelativeToRoot(new Path(fileName.toString()), EDGE_FILE_DEPTH)) {
+ edges.add(new Path(fileName.toString()));
+ }
+ }
+ });
+ for (Path edge: edges) {
+ addFlowEdge(newFlowGraph, edge.toString());
Review Comment:
Unfortunately it's due to the implementation of the file listener, it was originally written for Gobblin pipelines and thus it uses the Hadoop Path references. There are a lot of usages of Hadoop Path objects in GaaS that I find unnecessary, especially because doing so requires GaaS to stay on outdated versions of Java. Hence I try to avoid having new classes rely on the Hadoop Path library but for compatibility it seems hard to avoid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] AndyJiang99 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
AndyJiang99 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r975812541
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/GitFlowGraphMonitor.java:
##########
@@ -128,10 +131,11 @@ void processGitConfigChanges() throws GitAPIException, IOException {
static class GitFlowgraphComparator implements Comparator<DiffEntry>, Serializable {
public int compare(DiffEntry o1, DiffEntry o2) {
- Integer o1Depth = (o1.getNewPath() != null) ? (new Path(o1.getNewPath())).depth() : (new Path(o1.getOldPath())).depth();
- Integer o2Depth = (o2.getNewPath() != null) ? (new Path(o2.getNewPath())).depth() : (new Path(o2.getOldPath())).depth();
+ Integer o1Depth =
Review Comment:
Might be because I'm not too familiar with this, but could we change o1Depth and o2Depth to be a bit more descriptive? Not too sure what o1 and o2 stand for
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r970170238
##########
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/FsFlowGraphMonitorTest.java:
##########
@@ -0,0 +1,310 @@
+/*
+ * 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.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Optional;
+import com.google.common.io.Files;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import org.apache.commons.io.FileUtils;
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.flow.MultiHopFlowCompilerTest;
+import org.apache.gobblin.service.modules.flowgraph.BaseFlowGraph;
+import org.apache.gobblin.service.modules.flowgraph.DataNode;
+import org.apache.gobblin.service.modules.flowgraph.FlowEdge;
+import org.apache.gobblin.service.modules.flowgraph.FlowGraphConfigurationKeys;
+import org.apache.gobblin.service.modules.template_catalog.FSFlowTemplateCatalog;
+import org.apache.hadoop.fs.Path;
+import org.eclipse.jgit.transport.RefSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+
+public class FsFlowGraphMonitorTest {
+ private static final Logger logger = LoggerFactory.getLogger(GitFlowGraphMonitor.class);
Review Comment:
Oops 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] AndyJiang99 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
AndyJiang99 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r975719456
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -84,26 +84,27 @@ public BaseFlowGraphListener(Optional<? extends FSFlowTemplateCatalog> flowTempl
this.hoconFileExtensions = Sets.newHashSet(hoconFileExtensions.split(","));
try {
this.pullFileLoader = new PullFileLoader(folderPath,
- FileSystem.get(URI.create(ConfigurationKeys.LOCAL_FS_URI), new Configuration()),
- this.javaPropsExtensions, this.hoconFileExtensions);
+ FileSystem.get(URI.create(ConfigurationKeys.LOCAL_FS_URI), new Configuration()), this.javaPropsExtensions,
+ this.hoconFileExtensions);
} catch (IOException e) {
throw new RuntimeException("Could not create pull file loader", e);
}
}
+
/**
* Add a {@link DataNode} to the {@link FlowGraph}. The method uses the {@link FlowGraphConfigurationKeys#DATA_NODE_CLASS} config
* to instantiate a {@link DataNode} from the node config file.
* @param path of node to add
*/
- protected void addDataNode(String path) {
+ public void addDataNode(FlowGraph graph, String path) {
Review Comment:
Nit: indentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] ZihanLi58 commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r980554915
##########
gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/PathAlterationObserver.java:
##########
@@ -187,6 +189,13 @@ public void checkAndNotify()
for (final PathAlterationListener listener : listeners.values()) {
listener.onStop(this);
}
+
+ if (this.changeApplied) {
Review Comment:
Should this be added before the onStop method is called?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3548: [GOBBLIN-1696] Implement file based flowgraph that detects changes to the underlying…
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3548:
URL: https://github.com/apache/gobblin/pull/3548#discussion_r979698589
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/flowgraph/BaseFlowGraphHelper.java:
##########
@@ -277,32 +282,64 @@ private List<SpecExecutor> getSpecExecutors(Config edgeConfig) throws URISyntaxE
* @return the configuration object
* @throws IOException
*/
- protected Config loadNodeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadNodeFileWithOverrides(Path filePath)
+ throws IOException {
Config nodeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getNodeConfigWithOverrides(nodeConfig, filePath);
}
-
/**
* Load the edge file.
* @param filePath path of the edge file relative to the repository root
* @return the configuration object
* @throws IOException
*/
- protected Config loadEdgeFileWithOverrides(Path filePath) throws IOException {
+ protected Config loadEdgeFileWithOverrides(Path filePath)
+ throws IOException {
Config edgeConfig = this.pullFileLoader.loadPullFile(filePath, emptyConfig, false, false);
return getEdgeConfigWithOverrides(edgeConfig, filePath);
}
+ /**
+ * Loads the entire flowgraph from the path configured in {@link org.apache.gobblin.configuration.ConfigurationKeys.FLOWGRAPH_BASE_DIR }
+ * Expects nodes to be in the format of /flowGraphName/nodeA/nodeA.properties
+ * Expects edges to be in the format of /flowGraphName/nodeA/nodeB/edgeAB.properties
+ * The current flowgraph will be swapped atomically with the new flowgraph that is loaded
+ */
+ public void populateFlowGraphAtomically(AtomicReference<FlowGraph> flowGraphReference) {
Review Comment:
I was able to change the GitFlowGraphMonitor such that it also follows the same behavior as the FsFlowGraphMonitor, they will both update fully now and not perform partial updates. Seems like a more unified solution like you suggested
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org