Streamed segment download & untar with rate limiter to control disk usage #8753

jasperjiaguo · 2022-05-20T22:20:58Z

We found that during the segment push of large and daily-refresh dataset the disk usage is pretty spiky. It will basically max out the disk throughput (to a few GB/s for SSD), during which the query latency can be impacted. However, the average disk throughput is just at a few hundred MB/s level in minute granularity, which means if we can make better use of the disk the contention on bandwidth may be mitigated. Meanwhile the current offline->online process for an offline segment is basically download -> untar -> MMAP -> download -> untar -> …. This can potentially be improved in the following ways:

If we can combine download -> untar and directly write the untarred segments to disk we can save some unnecessary r/w.
If we can cap the download speed on top of limiting the number of segments download in parallel (see Set max number of parallel segment downloads per table in pinot-server #8694). We may may improve the query performance during refresh.

Test Table (2.7GB/segment, 114 segments):
CPU profiling using Top/ODP given the same downloading time

Disk profiling using PCP in 50ms granularity given the same downloading time:

IO size captured with PCP w.r.t. different buffer size:

Findings & Conclusions:

Experiment shows with Jack's change to control download parallelism we can control per table stream rate.
This change makes the disk write rate curve very constant even at 50ms granularity, as avoids disk saturation.
The streamed download-untar feature bypasses writing tarball to the disk, as will reduce the disk IO to 2/3 in the tested case.
ODP profiling shows the new download code has minimum CPU impact: RateLimiter: [0.583%] / FileDescriptor#sync: [0.583%], and top profiling also indicates similar/lower CPU utilization.
The rate limiter can also effectively control the network rx rate.

release notes
Configs:
pinot.server.instance.segment.stream.download.untar.rate.limit.bytes.per.sec
Key of streamed server segment download-untar rate limit
limit the rate to write download-untar stream to disk, in bytes
-1 for no disk write limit, 0 for limit the writing to min(untar, download) rate

pinot.server.instance.segment.stream.download.untar
Key of whether to use streamed server segment download-untar

codecov-commenter · 2022-05-21T08:14:57Z

Codecov Report

❌ Patch coverage is 23.25581% with 99 lines in your changes missing coverage. Please review.
✅ Project coverage is 69.61%. Comparing base (d1d2ddb) to head (2f0ad7e).
⚠️ Report is 5260 commits behind head on master.

Files with missing lines	Patch %	Lines
.../pinot/core/data/manager/BaseTableDataManager.java	17.07%	32 Missing and 2 partials ⚠️
...pinot/common/utils/fetcher/HttpSegmentFetcher.java	3.44%	28 Missing ⚠️
...ache/pinot/common/utils/TarGzCompressionUtils.java	15.78%	15 Missing and 1 partial ⚠️
...org/apache/pinot/common/utils/http/HttpClient.java	0.00%	11 Missing ⚠️
...ent/local/data/manager/TableDataManagerParams.java	70.00%	6 Missing ⚠️
...ot/common/utils/fetcher/SegmentFetcherFactory.java	0.00%	2 Missing ⚠️
...e/pinot/common/utils/FileUploadDownloadClient.java	0.00%	1 Missing ⚠️
...pinot/common/utils/fetcher/BaseSegmentFetcher.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #8753      +/-   ##
============================================
- Coverage     69.80%   69.61%   -0.20%     
- Complexity     4659     4680      +21     
============================================
  Files          1741     1805      +64     
  Lines         91476    94006    +2530     
  Branches      13677    13992     +315     
============================================
+ Hits          63854    65441    +1587     
- Misses        23199    24019     +820     
- Partials       4423     4546     +123

Flag	Coverage Δ
integration1	`26.30% <12.40%> (-0.78%)`	⬇️
integration2	`24.99% <12.40%> (-0.69%)`	⬇️
unittests1	`66.36% <21.25%> (+0.14%)`	⬆️
unittests2	`15.45% <0.00%> (+1.31%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java

pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcher.java

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

siddharthteotia · 2022-06-08T16:47:47Z

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

Curious why this change ?

Because we need to get the FileDescriptor from the FileOutputStream in streamCopyWithRateLimiter. Creating a FileOutputStream from outputFile should be safe here.

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

pinot-common/src/main/java/org/apache/pinot/common/utils/http/HttpClient.java

siddharthteotia · 2022-06-08T17:14:04Z

pinot-common/src/main/java/org/apache/pinot/common/utils/http/HttpClient.java

Is there a way to do a sanity check / validation when streaming untar finishes to ensure the download was correct / complete ?

Discussed offline. The untar library will fail on checksum and report IOException if files are corrupted during transmission

…efresh

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

jackjlli · 2022-06-09T20:43:39Z

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

+        fd.sync();
+      }
+    }
+    return count;


Log one more message to show that the copy is done.

This function is actually called per file. So I removed all the logging here as it will be too much. We already have similar logging in private File downloadAndStreamUntarRateLimit(String segmentName, SegmentZKMetadata zkMetadata, File tempRootDir, long maxDownloadRateInByte) which should be enough

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

...-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerParams.java

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

siddharthteotia · 2022-06-09T23:19:37Z

@jasperjiaguo - let's add a brief summary of the improvement numbers we saw from internal performance test on prod data.

jackjlli

Minor but LGTM. These can be addressed in later PR.

jackjlli · 2022-06-09T23:47:13Z

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

+  public static long copyWithRateLimiter(InputStream inputStream, FileOutputStream outputStream,
+      long maxStreamRateInByte)
+      throws IOException {
+    Preconditions.checkState(inputStream != null, "inputStream is null");


Preconditions.checkNotNull() might be better. Same for the following line.

jackjlli · 2022-06-09T23:47:50Z

pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/BaseSegmentFetcher.java

    });
  }

+  public File fetchUntarSegmentToLocalStreamed(URI uri, File dest, long rateLimit)


Rename the last param here. Same for other method.
Also, put a @Override annotation here.

jackjlli · 2022-06-09T23:55:03Z

pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcher.java

+  @Override
+  public File fetchUntarSegmentToLocalStreamed(URI downloadURI, File dest, long maxStreamRateInByte)
+      throws Exception {
+    // Create a RoundRobinURIProvider to round robin IP addresses when retry uploading. Otherwise may always try to


I think most of the logic here is the same as the one in fetchSegmentToLocal() method. It'd be good to check if we can extract the same logic to avoid duplicating the code.

jackjlli · 2022-06-09T23:58:22Z

...-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerParams.java

+  }
+
+  public TableDataManagerParams(InstanceDataManagerConfig instanceDataManagerConfig) {
+    _maxParallelSegmentDownloads = instanceDataManagerConfig.getMaxParallelSegmentDownloads();


Change it to sth like:
this(instanceDataManagerConfig.getMaxParallelSegmentDownloads(), instanceDataManagerConfig.isStreamSegmentDownloadUntar(), instanceDataManagerConfig.getStreamSegmentDownloadUntarRateLimit());

Also, rename the variable streamSegmentDownloadUntarRateLimit.

jackjlli · 2022-06-10T00:00:08Z

...-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerParams.java

+    return _streamSegmentDownloadUntarRateLimitBytesPerSec;
+  }
+
+  public void setStreamSegmentDownloadUntar(boolean streamSegmentDownloadUntar) {


You might not need all the setter methods here.

jackjlli · 2022-06-10T00:04:40Z

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

      throws Exception {
    File tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "-" + UUID.randomUUID());
    FileUtils.forceMkdir(tempRootDir);
+    if (_isStreamSegmentDownloadUntar && zkMetadata.getCrypterName() == null) {


Does that mean we cannot stream & untar if the segment is encrypted? If so, specify it to one comment in this method here.

Also, can this if statement be put into the try block, so that you don't have to duplicate the code for deleting tempRootDir.

jackjlli · 2022-06-10T00:11:17Z

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

+      FileUtils.moveDirectory(untaredSegDir, indexDir);
+      return indexDir;
+    } catch (Exception e) {
+      LOGGER.error("Failed to move segment: {} of table: {}", segmentName, _tableNameWithType);


It'd be good to print out what exact exception is thrown in this error message, if it doesn't get printed somewhere in the upper caller above.

jasperjiaguo · 2022-06-10T00:38:38Z

Minor but LGTM. These can be addressed in later PR.

Thanks, will address these in a follow-up PR

…sage (apache#8753) * Add streamed download untar with ratelimit for segment download and refresh * address comments * address comments * address comments * address comments * address comments * address comments * address comments

Jackie-Jiang · 2022-06-13T19:21:10Z

@jasperjiaguo This is great enhancement. I labeled the PR with release-notes, and can you please add a release-notes section to the PR description to list down the new config and metrics added in this PR? Also, can you please also help update the documentation?

jasperjiaguo · 2022-06-16T16:33:24Z

@jasperjiaguo This is great enhancement. I labeled the PR with release-notes, and can you please add a release-notes section to the PR description to list down the new config and metrics added in this PR? Also, can you please also help update the documentation?

@Jackie-Jiang added release notes for config. Will add them to documentation later.

jugomezv · 2022-08-01T23:14:10Z

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

+   * RateLimit limits the untar rate
+   * <p>For security reason, the untarred files must reside in the output directory.
+   */
+  public static List<File> untarRateLimit(InputStream inputStream, File outputDir, long rateLimit)


for the rateLimit argument, can we include the units in the argument name to make code more readable?

jugomezv · 2022-08-01T23:24:47Z

pinot-common/src/main/java/org/apache/pinot/common/utils/TarGzCompressionUtils.java

    }
  }
+
+  public static long copyLarge(InputStream inputStream, FileOutputStream outputStream, long rateLimit)


Should we assert that rateLimit is not NO_LIMIT?

jugomezv · 2022-08-01T23:31:48Z

pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcher.java

+          // Temporary exception
+          // 404 is treated as a temporary exception, as the downloadURI may be backed by multiple hosts,
+          // if singe host is down, can retry with another host.
+          _logger.warn("Got temporary error status code: {} while downloading segment from: {} to: {}", statusCode, uri,


is it useful to log the retry attempt here?

jugomezv · 2022-08-01T23:36:24Z

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

      throws Exception {
    File tempRootDir = getTmpSegmentDataDir("tmp-" + segmentName + "-" + UUID.randomUUID());
    FileUtils.forceMkdir(tempRootDir);
+    if (_isSegmentDownloadUntarStreamed && zkMetadata.getCrypterName() == null) {


so seems possible for me to configure streaming and encryption which is not a valid config but we do not log and just default to non streaming, should we log that in the else below?

jasperjiaguo changed the title ~~[WIP] Streamed segment download & untar with rate limiter to control disk usage feature~~ [WIP][Not ready for review] Streamed segment download & untar with rate limiter to control disk usage feature May 20, 2022

jasperjiaguo marked this pull request as draft May 20, 2022 23:17

jasperjiaguo force-pushed the limit_download_rate branch from f888915 to 713de1c Compare May 21, 2022 05:54

jasperjiaguo force-pushed the limit_download_rate branch 10 times, most recently from 9f7f905 to ef72a3e Compare May 27, 2022 06:02

jasperjiaguo force-pushed the limit_download_rate branch 6 times, most recently from d3ff77f to 9572b3e Compare June 7, 2022 16:26