-
Notifications
You must be signed in to change notification settings - Fork 409
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
317 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
...main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/web/IcebergObjectMapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * Copyright 2023 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
|
|
||
| package com.datastrato.gravitino.catalog.lakehouse.iceberg.web; | ||
|
|
||
| import com.datastrato.gravitino.json.JsonUtils; | ||
| import com.fasterxml.jackson.annotation.JsonAutoDetect; | ||
| import com.fasterxml.jackson.annotation.PropertyAccessor; | ||
| import com.fasterxml.jackson.databind.DeserializationFeature; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.databind.PropertyNamingStrategies; | ||
| import org.apache.iceberg.rest.RESTSerializers; | ||
|
|
||
| public class IcebergObjectMapper { | ||
| private static final ObjectMapper objectMapper = createIcebergObjectMapper(); | ||
|
|
||
| private IcebergObjectMapper() {} | ||
|
|
||
| private static ObjectMapper createIcebergObjectMapper() { | ||
| ObjectMapper mapper = JsonUtils.objectMapper(); | ||
| mapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); | ||
| mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); | ||
| mapper.setPropertyNamingStrategy(new PropertyNamingStrategies.KebabCaseStrategy()); | ||
| RESTSerializers.registerAll(mapper); | ||
| return mapper; | ||
| } | ||
|
|
||
| public static ObjectMapper getInstance() { | ||
| return objectMapper; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
...a/com/datastrato/gravitino/catalog/lakehouse/iceberg/web/metrics/EmptyMetricsStorage.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| /* | ||
| * Copyright 2023 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
|
|
||
| package com.datastrato.gravitino.catalog.lakehouse.iceberg.web.metrics; | ||
|
|
||
| import java.time.Instant; | ||
| import org.apache.iceberg.metrics.MetricsReport; | ||
|
|
||
| public class EmptyMetricsStorage implements IcebergMetricsStorage { | ||
| public void save(MetricsReport metricsReport) {} | ||
|
|
||
| public void close() {} | ||
|
|
||
| public void clean(Instant expireTime) {} | ||
| } |
24 changes: 24 additions & 0 deletions
24
...m/datastrato/gravitino/catalog/lakehouse/iceberg/web/metrics/IcebergMetricsFormatter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| /* | ||
| * Copyright 2023 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
|
|
||
| package com.datastrato.gravitino.catalog.lakehouse.iceberg.web.metrics; | ||
|
|
||
| import com.datastrato.gravitino.catalog.lakehouse.iceberg.web.IcebergObjectMapper; | ||
| import com.fasterxml.jackson.core.JsonProcessingException; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.apache.iceberg.metrics.MetricsReport; | ||
|
|
||
| public class IcebergMetricsFormatter { | ||
|
|
||
| private ObjectMapper icebergObjectMapper; | ||
|
|
||
| IcebergMetricsFormatter() { | ||
| this.icebergObjectMapper = IcebergObjectMapper.getInstance(); | ||
| } | ||
|
|
||
| public String toJson(MetricsReport metricsReport) throws JsonProcessingException { | ||
| return icebergObjectMapper.writeValueAsString(metricsReport); | ||
| } | ||
| } |
146 changes: 146 additions & 0 deletions
146
...com/datastrato/gravitino/catalog/lakehouse/iceberg/web/metrics/IcebergMetricsManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| /* | ||
| * Copyright 2023 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
|
|
||
| package com.datastrato.gravitino.catalog.lakehouse.iceberg.web.metrics; | ||
|
|
||
| import com.datastrato.gravitino.catalog.lakehouse.iceberg.IcebergConfig; | ||
| import com.datastrato.gravitino.catalog.lakehouse.iceberg.ops.IcebergTableOps; | ||
| import com.datastrato.gravitino.catalog.lakehouse.iceberg.web.IcebergRestUtils; | ||
| import com.google.common.util.concurrent.ThreadFactoryBuilder; | ||
| import java.io.IOException; | ||
| import java.time.Instant; | ||
| import java.util.concurrent.BlockingQueue; | ||
| import java.util.concurrent.LinkedBlockingQueue; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
| import java.util.concurrent.ScheduledThreadPoolExecutor; | ||
| import java.util.concurrent.TimeUnit; | ||
| import org.apache.iceberg.metrics.MetricsReport; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public class IcebergMetricsManager { | ||
| private static final Logger LOG = LoggerFactory.getLogger(IcebergTableOps.class); | ||
| private IcebergMetricsFormatter icebergMetricsFormatter; | ||
|
|
||
| enum MetricsStorageType { | ||
| EMPTY | ||
| }; | ||
|
|
||
| private IcebergMetricsStorage storage; | ||
| private boolean asyncMode = true; | ||
| private BlockingQueue<MetricsReport> queue; | ||
| private Thread metricsWriterThread; | ||
| private int retainDays; | ||
| private ScheduledExecutorService scheduledThreadPoolExecutor; | ||
| public static final String ICEBERG_METRICS_STORAGE = "metrics-storage"; | ||
| public static final String ICEBERG_METRICS_STORAGE_RETAIN_DAYS = "metrics-storage-retain-days"; | ||
|
|
||
| public IcebergMetricsManager(IcebergConfig icebergConfig) { | ||
| icebergMetricsFormatter = new IcebergMetricsFormatter(); | ||
| String metricsStorage = icebergConfig.get(IcebergConfig.ICEBERG_METRICS_STORAGE); | ||
| MetricsStorageType type = MetricsStorageType.valueOf(metricsStorage.toUpperCase()); | ||
| LOG.info("Iceberg metrics storage: {}", type); | ||
| switch (type) { | ||
| case EMPTY: | ||
| storage = new EmptyMetricsStorage(); | ||
| break; | ||
| default: | ||
| throw new RuntimeException("Unknown Iceberg metrics storage: " + type); | ||
| } | ||
|
|
||
| retainDays = icebergConfig.get(IcebergConfig.ICEBERG_METRICS_STORAGE_RETAIN_DAYS); | ||
| if (retainDays > 0) { | ||
| scheduledThreadPoolExecutor = | ||
| new ScheduledThreadPoolExecutor( | ||
| 1, | ||
| new ThreadFactoryBuilder() | ||
| .setDaemon(true) | ||
| .setNameFormat("Iceberg-metrics-cleaner") | ||
| .setUncaughtExceptionHandler( | ||
| (t, e) -> LOG.error("Uncaught exception in thread {}", t, e)) | ||
| .build()); | ||
| } | ||
|
|
||
| if (asyncMode) { | ||
| queue = new LinkedBlockingQueue(1000); | ||
| metricsWriterThread = | ||
| new Thread( | ||
| () -> { | ||
| while (Thread.currentThread().isInterrupted() == false) { | ||
| MetricsReport metricsReport; | ||
| try { | ||
| metricsReport = queue.poll(1, TimeUnit.MINUTES); | ||
| } catch (InterruptedException e) { | ||
| LOG.warn("Iceberg Metrics writer thread is interrupted"); | ||
| return; | ||
| } | ||
| if (metricsReport != null) { | ||
| doSave(metricsReport); | ||
| } | ||
| } | ||
| }); | ||
| metricsWriterThread.setDaemon(true); | ||
| } | ||
| } | ||
|
|
||
| public void start() { | ||
| if (metricsWriterThread != null) { | ||
| metricsWriterThread.start(); | ||
| } | ||
|
|
||
| if (scheduledThreadPoolExecutor != null) { | ||
| scheduledThreadPoolExecutor.scheduleAtFixedRate( | ||
| () -> { | ||
| Instant now = Instant.now(); | ||
| Instant expireTime = IcebergRestUtils.calculateNewTimestamp(now, -24 * retainDays); | ||
| LOG.info("Try clean Iceberg expired metrics, {}", expireTime); | ||
| try { | ||
| storage.clean(expireTime); | ||
| } catch (Exception e) { | ||
| LOG.warn("Clean Iceberg metrics failed,", e); | ||
| } | ||
| }, | ||
| 0, | ||
| 1, | ||
| TimeUnit.HOURS); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private void doSave(MetricsReport metricsReport) { | ||
| try { | ||
| storage.save(metricsReport); | ||
| } catch (Exception e) { | ||
| LOG.warn("Write Iceberg metrics failed,", e); | ||
| } | ||
| } | ||
|
|
||
| public void save(MetricsReport metricsReport) { | ||
| if (asyncMode) { | ||
| if (queue.offer(metricsReport) == false) { | ||
| LOG.warn("Iceberg metrics queue is full, drop metrics report"); | ||
| } | ||
| return; | ||
| } | ||
| doSave(metricsReport); | ||
| } | ||
|
|
||
| public void close() { | ||
| if (metricsWriterThread != null) { | ||
| metricsWriterThread.interrupt(); | ||
| // no need to join metrics writer thread | ||
| } | ||
|
|
||
| if (scheduledThreadPoolExecutor != null) { | ||
| scheduledThreadPoolExecutor.shutdownNow(); | ||
| } | ||
|
|
||
| try { | ||
| storage.close(); | ||
| } catch (IOException e) { | ||
| LOG.warn("Close Iceberg metrics storage failed", e); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.