[
https://issues.apache.org/jira/browse/DRILL-5270?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16474316#comment-16474316
]
ASF GitHub Bot commented on DRILL-5270:
---------------------------------------
arina-ielchiieva commented on a change in pull request #1250: DRILL-5270:
Improve loading of profiles listing in the WebUI
URL: https://github.com/apache/drill/pull/1250#discussion_r187984321
##########
File path:
exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/LocalPersistentStore.java
##########
@@ -1,220 +1,377 @@
-/*
- * 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.drill.exec.store.sys.store;
-
-import static org.apache.drill.exec.ExecConstants.DRILL_SYS_FILE_SUFFIX;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import javax.annotation.Nullable;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.drill.common.collections.ImmutableEntry;
-import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.exec.store.dfs.DrillFileSystem;
-import org.apache.drill.exec.util.DrillFileSystemUtil;
-import org.apache.drill.exec.store.sys.BasePersistentStore;
-import org.apache.drill.exec.store.sys.PersistentStoreConfig;
-import org.apache.drill.exec.store.sys.PersistentStoreMode;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import org.apache.hadoop.fs.PathFilter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LocalPersistentStore<V> extends BasePersistentStore<V> {
- private static final Logger logger =
LoggerFactory.getLogger(LocalPersistentStore.class);
-
- private final Path basePath;
- private final PersistentStoreConfig<V> config;
- private final DrillFileSystem fs;
-
- public LocalPersistentStore(DrillFileSystem fs, Path base,
PersistentStoreConfig<V> config) {
- this.basePath = new Path(base, config.getName());
- this.config = config;
- this.fs = fs;
- try {
- mkdirs(getBasePath());
- } catch (IOException e) {
- throw new RuntimeException("Failure setting pstore configuration path.");
- }
- }
-
- protected Path getBasePath() {
- return basePath;
- }
-
- @Override
- public PersistentStoreMode getMode() {
- return PersistentStoreMode.PERSISTENT;
- }
-
- private void mkdirs(Path path) throws IOException {
- fs.mkdirs(path);
- }
-
- public static Path getLogDir() {
- String drillLogDir = System.getenv("DRILL_LOG_DIR");
- if (drillLogDir == null) {
- drillLogDir = System.getProperty("drill.log.dir");
- }
- if (drillLogDir == null) {
- drillLogDir = "/var/log/drill";
- }
- return new Path(new File(drillLogDir).getAbsoluteFile().toURI());
- }
-
- public static DrillFileSystem getFileSystem(DrillConfig config, Path root)
throws IOException {
- Path blobRoot = root == null ? getLogDir() : root;
- Configuration fsConf = new Configuration();
- if (blobRoot.toUri().getScheme() != null) {
- fsConf.set(FileSystem.FS_DEFAULT_NAME_KEY, blobRoot.toUri().toString());
- }
-
-
- DrillFileSystem fs = new DrillFileSystem(fsConf);
- fs.mkdirs(blobRoot);
- return fs;
- }
-
- @Override
- public Iterator<Map.Entry<String, V>> getRange(int skip, int take) {
- try {
- // list only files with sys file suffix
- PathFilter sysFileSuffixFilter = new PathFilter() {
- @Override
- public boolean accept(Path path) {
- return path.getName().endsWith(DRILL_SYS_FILE_SUFFIX);
- }
- };
-
- List<FileStatus> fileStatuses = DrillFileSystemUtil.listFiles(fs,
basePath, false, sysFileSuffixFilter);
- if (fileStatuses.isEmpty()) {
- return Collections.emptyIterator();
- }
-
- List<String> files = Lists.newArrayList();
- for (FileStatus stat : fileStatuses) {
- String s = stat.getPath().getName();
- files.add(s.substring(0, s.length() - DRILL_SYS_FILE_SUFFIX.length()));
- }
-
- Collections.sort(files);
-
- return Iterables.transform(Iterables.limit(Iterables.skip(files, skip),
take), new Function<String, Entry<String, V>>() {
- @Nullable
- @Override
- public Entry<String, V> apply(String key) {
- return new ImmutableEntry<>(key, get(key));
- }
- }).iterator();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- private Path makePath(String name) {
- Preconditions.checkArgument(
- !name.contains("/") &&
- !name.contains(":") &&
- !name.contains(".."));
- return new Path(basePath, name + DRILL_SYS_FILE_SUFFIX);
- }
-
- @Override
- public boolean contains(String key) {
- try {
- return fs.exists(makePath(key));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public V get(String key) {
- try {
- Path path = makePath(key);
- if (!fs.exists(path)) {
- return null;
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- final Path path = makePath(key);
- try (InputStream is = fs.open(path)) {
- return config.getSerializer().deserialize(IOUtils.toByteArray(is));
- } catch (IOException e) {
- throw new RuntimeException("Unable to deserialize \"" + path + "\"", e);
- }
- }
-
- @Override
- public void put(String key, V value) {
- try (OutputStream os = fs.create(makePath(key))) {
- IOUtils.write(config.getSerializer().serialize(value), os);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public boolean putIfAbsent(String key, V value) {
- try {
- Path p = makePath(key);
- if (fs.exists(p)) {
- return false;
- } else {
- try (OutputStream os = fs.create(makePath(key))) {
- IOUtils.write(config.getSerializer().serialize(value), os);
- }
- return true;
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void delete(String key) {
- try {
- fs.delete(makePath(key), false);
- } catch (IOException e) {
- logger.error("Unable to delete data from storage.", e);
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void close() {
- }
-}
+/*
+ * 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.drill.exec.store.sys.store;
+
+import static org.apache.drill.exec.ExecConstants.DRILL_SYS_FILE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.Nullable;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.drill.common.collections.ImmutableEntry;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.sys.BasePersistentStore;
+import org.apache.drill.exec.store.sys.PersistentStoreConfig;
+import org.apache.drill.exec.store.sys.PersistentStoreMode;
+import org.apache.drill.exec.util.DrillFileSystemUtil;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Iterables;
+
+public class LocalPersistentStore<V> extends BasePersistentStore<V> {
+ private static final Logger logger =
LoggerFactory.getLogger(LocalPersistentStore.class);
+
+ //Provides a threshold above which we report an event's time
+ private static final long RESPONSE_TIME_THRESHOLD_MSEC = 2000L;
+ private static final String ARCHIVE_LOCATION = "archived";
+
+ private static final int DRILL_SYS_FILE_EXT_SIZE =
DRILL_SYS_FILE_SUFFIX.length();
+ private final Path basePath;
+ private final PersistentStoreConfig<V> config;
+ private final DrillFileSystem fs;
+ private int version = -1;
+ private Function<String, Entry<String, V>> transformer;
+
+ private ProfileSet profilesSet;
+ private PathFilter sysFileSuffixFilter;
+// private String mostRecentProfile;
+ private long basePathLastModified;
+ private long lastKnownFileCount;
+ private int maxSetCapacity;
+ private Stopwatch listAndBuildWatch;
+ private Stopwatch transformWatch;
+
+ private boolean enableArchiving;
+ private Path archivePath;
+ private ProfileSet pendingArchivalSet;
+ private int archivalThreshold;
+ private int archivalRate;
+ private Iterable<Entry<String, V>> iterableProfileSet;
+ private Stopwatch archiveWatch;
+
+ public LocalPersistentStore(DrillFileSystem fs, Path base,
PersistentStoreConfig<V> config, DrillConfig drillConfig) {
+ super();
+ this.basePath = new Path(base, config.getName());
+ this.config = config;
+ this.fs = fs;
+ //MRU Profile Cache
+ this.profilesSet = new
ProfileSet(drillConfig.getInt(ExecConstants.HTTP_MAX_PROFILES));
+ this.basePathLastModified = 0L;
+ this.lastKnownFileCount = 0L;
+ this.sysFileSuffixFilter = new DrillSysFilePathFilter();
+ //this.mostRecentProfile = null;
+
+ //Initializing for archiving
+ this.enableArchiving =
drillConfig.getBoolean(ExecConstants.PROFILES_STORE_ARCHIVE_ENABLED);
//(maxStoreCapacity > 0); //Implicitly infer
+ if (enableArchiving == true ) {
+ this.archivalThreshold =
drillConfig.getInt(ExecConstants.PROFILES_STORE_CAPACITY);
+ this.archivalRate =
drillConfig.getInt(ExecConstants.PROFILES_STORE_ARCHIVE_RATE);
+ this.pendingArchivalSet = new ProfileSet(archivalRate);
+ this.archivePath = new Path(basePath, ARCHIVE_LOCATION);
+ }
+
+ // Timing
+ this.listAndBuildWatch = Stopwatch.createUnstarted();
+ this.transformWatch = Stopwatch.createUnstarted();
+ this.archiveWatch = Stopwatch.createUnstarted();
+
+ // One time transformer function instantiation
+ this.transformer = new Function<String, Entry<String, V>>() {
+ @Nullable
+ @Override
+ public Entry<String, V> apply(String key) {
+ return new ImmutableEntry<>(key, get(key));
+ }
+ };
+
+ //Base Dir
+ try {
+ if (!mkdirs(basePath)) {
+ version++;
+ }
+ //Creating Archive if required
+ if (enableArchiving) {
+ try {
+ if (!fs.exists(archivePath)) {
+ mkdirs(archivePath);
+ }
+ } catch (IOException e) {
+ logger.warn("Disabling profile archiving due to failure in creating
profile archive {} : {}", archivePath, e);
+ this.enableArchiving = false;
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Failure setting pstore configuration path.");
+ }
+ }
+
+ protected Path getBasePath() {
+ return basePath;
+ }
+
+ @Override
+ public PersistentStoreMode getMode() {
+ return PersistentStoreMode.PERSISTENT;
+ }
+
+ private boolean mkdirs(Path path) throws IOException {
+ return fs.mkdirs(path);
+ }
+
+ public static Path getLogDir() {
+ String drillLogDir = System.getenv("DRILL_LOG_DIR");
+ if (drillLogDir == null) {
+ drillLogDir = System.getProperty("drill.log.dir");
+ }
+ if (drillLogDir == null) {
+ drillLogDir = "/var/log/drill";
+ }
+ return new Path(new File(drillLogDir).getAbsoluteFile().toURI());
+ }
+
+ public static DrillFileSystem getFileSystem(DrillConfig config, Path root)
throws IOException {
+ Path blobRoot = root == null ? getLogDir() : root;
+ Configuration fsConf = new Configuration();
+ if (blobRoot.toUri().getScheme() != null) {
+ fsConf.set(FileSystem.FS_DEFAULT_NAME_KEY, blobRoot.toUri().toString());
+ }
+
+
+ DrillFileSystem fs = new DrillFileSystem(fsConf);
+ fs.mkdirs(blobRoot);
+ return fs;
+ }
+
+ @Override
+ public Iterator<Map.Entry<String, V>> getRange(int skip, int take) {
+ //Marking currently seen modification time
+ long currBasePathModified = 0L;
+ try {
+ currBasePathModified = fs.getFileStatus(basePath).getModificationTime();
+ } catch (IOException e) {
+ logger.error("Failed to get FileStatus for {}", basePath, e);
+ throw new RuntimeException(e);
+ }
+
+ //No need to acquire lock since incoming requests are synchronized
Review comment:
Could you please point out to the code which proves this statement?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Improve loading of profiles listing in the WebUI
> ------------------------------------------------
>
> Key: DRILL-5270
> URL: https://issues.apache.org/jira/browse/DRILL-5270
> Project: Apache Drill
> Issue Type: Improvement
> Components: Web Server
> Affects Versions: 1.9.0
> Reporter: Kunal Khatua
> Assignee: Kunal Khatua
> Priority: Major
> Fix For: 1.14.0
>
>
> Currently, as the number of profiles increase, we reload the same list of
> profiles from the FS.
> An ideal improvement would be to detect if there are any new profiles and
> only reload from the disk then. Otherwise, a cached list is sufficient.
> For a directory of 280K profiles, the load time is close to 6 seconds on a 32
> core server. With the caching, we can get it down to as much as a few
> milliseconds.
> To render the cache as invalid, we inspect the last modified time of the
> directory to confirm whether a reload is needed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
