zhoujinsong commented on code in PR #3429: URL: https://github.com/apache/amoro/pull/3429#discussion_r1944303166
########## amoro-ams/src/main/java/org/apache/amoro/server/dashboard/OverviewManager.java: ########## @@ -0,0 +1,274 @@ +/* + * 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.amoro.server.dashboard; + +import org.apache.amoro.config.Configurations; +import org.apache.amoro.server.AmoroManagementConf; +import org.apache.amoro.server.dashboard.model.OverviewDataSizeItem; +import org.apache.amoro.server.dashboard.model.OverviewResourceUsageItem; +import org.apache.amoro.server.dashboard.model.OverviewTopTableItem; +import org.apache.amoro.server.optimizing.OptimizingStatus; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.TableRuntimeMeta; +import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper; +import org.apache.amoro.server.persistence.mapper.OptimizerMapper; +import org.apache.amoro.server.persistence.mapper.TableMetaMapper; +import org.apache.amoro.server.resource.OptimizerInstance; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +public class OverviewManager extends PersistentBase { + + public static final String STATUS_PENDING = "Pending"; + public static final String STATUS_PLANING = "Planing"; + public static final String STATUS_EXECUTING = "Executing"; + public static final String STATUS_IDLE = "Idle"; + public static final String STATUS_COMMITTING = "Committing"; + + private static final Logger LOG = LoggerFactory.getLogger(OverviewManager.class); + private final List<OverviewTopTableItem> allTopTableItem = new ArrayList<>(); + private final Map<String, Long> optimizingStatusCountMap = new ConcurrentHashMap<>(); + private final ConcurrentLinkedDeque<OverviewResourceUsageItem> resourceUsageHistory = + new ConcurrentLinkedDeque<>(); + private final ConcurrentLinkedDeque<OverviewDataSizeItem> dataSizeHistory = + new ConcurrentLinkedDeque<>(); + private final AtomicInteger totalCatalog = new AtomicInteger(); + private final AtomicLong totalDataSize = new AtomicLong(); + private final AtomicInteger totalTableCount = new AtomicInteger(); + private final AtomicInteger totalCpu = new AtomicInteger(); + private final AtomicLong totalMemory = new AtomicLong(); + + private final int maxRecordCount; + + public OverviewManager(Configurations serverConfigs) { + this( + serverConfigs.getInteger(AmoroManagementConf.OVERVIEW_CACHE_MAX_SIZE), + serverConfigs.get(AmoroManagementConf.OVERVIEW_CACHE_REFRESH_INTERVAL)); + } + + @VisibleForTesting + public OverviewManager(int maxRecordCount, Duration refreshInterval) { + this.maxRecordCount = maxRecordCount; + ScheduledExecutorService overviewUpdaterScheduler = Review Comment: `overviewRefreshScheduler ` may be a proper name, and the thread name pattern can be `overview-refresh-scheduler-%d`. ########## amoro-ams/src/main/java/org/apache/amoro/server/dashboard/OverviewManager.java: ########## @@ -0,0 +1,274 @@ +/* + * 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.amoro.server.dashboard; + +import org.apache.amoro.config.Configurations; +import org.apache.amoro.server.AmoroManagementConf; +import org.apache.amoro.server.dashboard.model.OverviewDataSizeItem; +import org.apache.amoro.server.dashboard.model.OverviewResourceUsageItem; +import org.apache.amoro.server.dashboard.model.OverviewTopTableItem; +import org.apache.amoro.server.optimizing.OptimizingStatus; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.TableRuntimeMeta; +import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper; +import org.apache.amoro.server.persistence.mapper.OptimizerMapper; +import org.apache.amoro.server.persistence.mapper.TableMetaMapper; +import org.apache.amoro.server.resource.OptimizerInstance; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +public class OverviewManager extends PersistentBase { + + public static final String STATUS_PENDING = "Pending"; + public static final String STATUS_PLANING = "Planing"; + public static final String STATUS_EXECUTING = "Executing"; + public static final String STATUS_IDLE = "Idle"; + public static final String STATUS_COMMITTING = "Committing"; + + private static final Logger LOG = LoggerFactory.getLogger(OverviewManager.class); + private final List<OverviewTopTableItem> allTopTableItem = new ArrayList<>(); + private final Map<String, Long> optimizingStatusCountMap = new ConcurrentHashMap<>(); + private final ConcurrentLinkedDeque<OverviewResourceUsageItem> resourceUsageHistory = + new ConcurrentLinkedDeque<>(); + private final ConcurrentLinkedDeque<OverviewDataSizeItem> dataSizeHistory = + new ConcurrentLinkedDeque<>(); + private final AtomicInteger totalCatalog = new AtomicInteger(); + private final AtomicLong totalDataSize = new AtomicLong(); + private final AtomicInteger totalTableCount = new AtomicInteger(); + private final AtomicInteger totalCpu = new AtomicInteger(); + private final AtomicLong totalMemory = new AtomicLong(); + + private final int maxRecordCount; + + public OverviewManager(Configurations serverConfigs) { + this( + serverConfigs.getInteger(AmoroManagementConf.OVERVIEW_CACHE_MAX_SIZE), + serverConfigs.get(AmoroManagementConf.OVERVIEW_CACHE_REFRESH_INTERVAL)); + } + + @VisibleForTesting + public OverviewManager(int maxRecordCount, Duration refreshInterval) { + this.maxRecordCount = maxRecordCount; + ScheduledExecutorService overviewUpdaterScheduler = + Executors.newSingleThreadScheduledExecutor( + new ThreadFactoryBuilder() + .setNameFormat("overview-updater-scheduler-%d") + .setDaemon(true) + .build()); + resetStatusMap(); + + if (refreshInterval.toMillis() > 0) { + overviewUpdaterScheduler.scheduleAtFixedRate( + this::refresh, 1000L, refreshInterval.toMillis(), TimeUnit.MILLISECONDS); + } + } + + public List<OverviewTopTableItem> getAllTopTableItem() { + return ImmutableList.copyOf(allTopTableItem); + } + + public int getTotalCatalog() { + return totalCatalog.get(); + } + + public int getTotalTableCount() { + return totalTableCount.get(); + } + + public long getTotalDataSize() { + return totalDataSize.get(); + } + + public int getTotalCpu() { + return totalCpu.get(); + } + + public long getTotalMemory() { + return totalMemory.get(); + } + + public List<OverviewResourceUsageItem> getResourceUsageHistory(long startTime) { + return resourceUsageHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public List<OverviewDataSizeItem> getDataSizeHistory(long startTime) { + return dataSizeHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public Map<String, Long> getOptimizingStatus() { + return optimizingStatusCountMap; + } + + @VisibleForTesting + public void refresh() { + long start = System.currentTimeMillis(); + LOG.info("Updating overview cache"); + try { + refreshTableCache(start); + refreshResourceUsage(start); + + } catch (Exception e) { + LOG.error("OverviewRefresher error", e); + } finally { + long end = System.currentTimeMillis(); + LOG.info("Refresher overview cache took {} ms.", end - start); Review Comment: ```suggestion LOG.info("Refreshed overview cache in {} ms", end - start); ``` ########## amoro-ams/src/main/java/org/apache/amoro/server/dashboard/OverviewManager.java: ########## @@ -0,0 +1,274 @@ +/* + * 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.amoro.server.dashboard; + +import org.apache.amoro.config.Configurations; +import org.apache.amoro.server.AmoroManagementConf; +import org.apache.amoro.server.dashboard.model.OverviewDataSizeItem; +import org.apache.amoro.server.dashboard.model.OverviewResourceUsageItem; +import org.apache.amoro.server.dashboard.model.OverviewTopTableItem; +import org.apache.amoro.server.optimizing.OptimizingStatus; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.TableRuntimeMeta; +import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper; +import org.apache.amoro.server.persistence.mapper.OptimizerMapper; +import org.apache.amoro.server.persistence.mapper.TableMetaMapper; +import org.apache.amoro.server.resource.OptimizerInstance; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +public class OverviewManager extends PersistentBase { + + public static final String STATUS_PENDING = "Pending"; + public static final String STATUS_PLANING = "Planing"; + public static final String STATUS_EXECUTING = "Executing"; + public static final String STATUS_IDLE = "Idle"; + public static final String STATUS_COMMITTING = "Committing"; + + private static final Logger LOG = LoggerFactory.getLogger(OverviewManager.class); + private final List<OverviewTopTableItem> allTopTableItem = new ArrayList<>(); + private final Map<String, Long> optimizingStatusCountMap = new ConcurrentHashMap<>(); + private final ConcurrentLinkedDeque<OverviewResourceUsageItem> resourceUsageHistory = + new ConcurrentLinkedDeque<>(); + private final ConcurrentLinkedDeque<OverviewDataSizeItem> dataSizeHistory = + new ConcurrentLinkedDeque<>(); + private final AtomicInteger totalCatalog = new AtomicInteger(); + private final AtomicLong totalDataSize = new AtomicLong(); + private final AtomicInteger totalTableCount = new AtomicInteger(); + private final AtomicInteger totalCpu = new AtomicInteger(); + private final AtomicLong totalMemory = new AtomicLong(); + + private final int maxRecordCount; + + public OverviewManager(Configurations serverConfigs) { + this( + serverConfigs.getInteger(AmoroManagementConf.OVERVIEW_CACHE_MAX_SIZE), + serverConfigs.get(AmoroManagementConf.OVERVIEW_CACHE_REFRESH_INTERVAL)); + } + + @VisibleForTesting + public OverviewManager(int maxRecordCount, Duration refreshInterval) { + this.maxRecordCount = maxRecordCount; + ScheduledExecutorService overviewUpdaterScheduler = + Executors.newSingleThreadScheduledExecutor( + new ThreadFactoryBuilder() + .setNameFormat("overview-updater-scheduler-%d") + .setDaemon(true) + .build()); + resetStatusMap(); + + if (refreshInterval.toMillis() > 0) { + overviewUpdaterScheduler.scheduleAtFixedRate( + this::refresh, 1000L, refreshInterval.toMillis(), TimeUnit.MILLISECONDS); + } + } + + public List<OverviewTopTableItem> getAllTopTableItem() { + return ImmutableList.copyOf(allTopTableItem); + } + + public int getTotalCatalog() { + return totalCatalog.get(); + } + + public int getTotalTableCount() { + return totalTableCount.get(); + } + + public long getTotalDataSize() { + return totalDataSize.get(); + } + + public int getTotalCpu() { + return totalCpu.get(); + } + + public long getTotalMemory() { + return totalMemory.get(); + } + + public List<OverviewResourceUsageItem> getResourceUsageHistory(long startTime) { + return resourceUsageHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public List<OverviewDataSizeItem> getDataSizeHistory(long startTime) { + return dataSizeHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public Map<String, Long> getOptimizingStatus() { + return optimizingStatusCountMap; + } + + @VisibleForTesting + public void refresh() { + long start = System.currentTimeMillis(); + LOG.info("Updating overview cache"); Review Comment: ```suggestion LOG.info("Refreshing overview cache"); ``` ########## amoro-ams/src/main/java/org/apache/amoro/server/dashboard/OverviewManager.java: ########## @@ -0,0 +1,274 @@ +/* + * 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.amoro.server.dashboard; + +import org.apache.amoro.config.Configurations; +import org.apache.amoro.server.AmoroManagementConf; +import org.apache.amoro.server.dashboard.model.OverviewDataSizeItem; +import org.apache.amoro.server.dashboard.model.OverviewResourceUsageItem; +import org.apache.amoro.server.dashboard.model.OverviewTopTableItem; +import org.apache.amoro.server.optimizing.OptimizingStatus; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.TableRuntimeMeta; +import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper; +import org.apache.amoro.server.persistence.mapper.OptimizerMapper; +import org.apache.amoro.server.persistence.mapper.TableMetaMapper; +import org.apache.amoro.server.resource.OptimizerInstance; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +public class OverviewManager extends PersistentBase { + + public static final String STATUS_PENDING = "Pending"; + public static final String STATUS_PLANING = "Planing"; + public static final String STATUS_EXECUTING = "Executing"; + public static final String STATUS_IDLE = "Idle"; + public static final String STATUS_COMMITTING = "Committing"; + + private static final Logger LOG = LoggerFactory.getLogger(OverviewManager.class); + private final List<OverviewTopTableItem> allTopTableItem = new ArrayList<>(); + private final Map<String, Long> optimizingStatusCountMap = new ConcurrentHashMap<>(); + private final ConcurrentLinkedDeque<OverviewResourceUsageItem> resourceUsageHistory = + new ConcurrentLinkedDeque<>(); + private final ConcurrentLinkedDeque<OverviewDataSizeItem> dataSizeHistory = + new ConcurrentLinkedDeque<>(); + private final AtomicInteger totalCatalog = new AtomicInteger(); + private final AtomicLong totalDataSize = new AtomicLong(); + private final AtomicInteger totalTableCount = new AtomicInteger(); + private final AtomicInteger totalCpu = new AtomicInteger(); + private final AtomicLong totalMemory = new AtomicLong(); + + private final int maxRecordCount; + + public OverviewManager(Configurations serverConfigs) { + this( + serverConfigs.getInteger(AmoroManagementConf.OVERVIEW_CACHE_MAX_SIZE), + serverConfigs.get(AmoroManagementConf.OVERVIEW_CACHE_REFRESH_INTERVAL)); + } + + @VisibleForTesting + public OverviewManager(int maxRecordCount, Duration refreshInterval) { + this.maxRecordCount = maxRecordCount; + ScheduledExecutorService overviewUpdaterScheduler = + Executors.newSingleThreadScheduledExecutor( + new ThreadFactoryBuilder() + .setNameFormat("overview-updater-scheduler-%d") + .setDaemon(true) + .build()); + resetStatusMap(); + + if (refreshInterval.toMillis() > 0) { + overviewUpdaterScheduler.scheduleAtFixedRate( + this::refresh, 1000L, refreshInterval.toMillis(), TimeUnit.MILLISECONDS); + } + } + + public List<OverviewTopTableItem> getAllTopTableItem() { + return ImmutableList.copyOf(allTopTableItem); + } + + public int getTotalCatalog() { + return totalCatalog.get(); + } + + public int getTotalTableCount() { + return totalTableCount.get(); + } + + public long getTotalDataSize() { + return totalDataSize.get(); + } + + public int getTotalCpu() { + return totalCpu.get(); + } + + public long getTotalMemory() { + return totalMemory.get(); + } + + public List<OverviewResourceUsageItem> getResourceUsageHistory(long startTime) { + return resourceUsageHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public List<OverviewDataSizeItem> getDataSizeHistory(long startTime) { + return dataSizeHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public Map<String, Long> getOptimizingStatus() { + return optimizingStatusCountMap; + } + + @VisibleForTesting + public void refresh() { + long start = System.currentTimeMillis(); + LOG.info("Updating overview cache"); + try { + refreshTableCache(start); + refreshResourceUsage(start); + + } catch (Exception e) { + LOG.error("OverviewRefresher error", e); Review Comment: ```suggestion LOG.error("Refreshed overview cache failed", e); ``` ########## amoro-ams/src/main/java/org/apache/amoro/server/dashboard/OverviewManager.java: ########## @@ -0,0 +1,274 @@ +/* + * 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.amoro.server.dashboard; + +import org.apache.amoro.config.Configurations; +import org.apache.amoro.server.AmoroManagementConf; +import org.apache.amoro.server.dashboard.model.OverviewDataSizeItem; +import org.apache.amoro.server.dashboard.model.OverviewResourceUsageItem; +import org.apache.amoro.server.dashboard.model.OverviewTopTableItem; +import org.apache.amoro.server.optimizing.OptimizingStatus; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.TableRuntimeMeta; +import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper; +import org.apache.amoro.server.persistence.mapper.OptimizerMapper; +import org.apache.amoro.server.persistence.mapper.TableMetaMapper; +import org.apache.amoro.server.resource.OptimizerInstance; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableList; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; +import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +public class OverviewManager extends PersistentBase { + + public static final String STATUS_PENDING = "Pending"; + public static final String STATUS_PLANING = "Planing"; + public static final String STATUS_EXECUTING = "Executing"; + public static final String STATUS_IDLE = "Idle"; + public static final String STATUS_COMMITTING = "Committing"; + + private static final Logger LOG = LoggerFactory.getLogger(OverviewManager.class); + private final List<OverviewTopTableItem> allTopTableItem = new ArrayList<>(); + private final Map<String, Long> optimizingStatusCountMap = new ConcurrentHashMap<>(); + private final ConcurrentLinkedDeque<OverviewResourceUsageItem> resourceUsageHistory = + new ConcurrentLinkedDeque<>(); + private final ConcurrentLinkedDeque<OverviewDataSizeItem> dataSizeHistory = + new ConcurrentLinkedDeque<>(); + private final AtomicInteger totalCatalog = new AtomicInteger(); + private final AtomicLong totalDataSize = new AtomicLong(); + private final AtomicInteger totalTableCount = new AtomicInteger(); + private final AtomicInteger totalCpu = new AtomicInteger(); + private final AtomicLong totalMemory = new AtomicLong(); + + private final int maxRecordCount; + + public OverviewManager(Configurations serverConfigs) { + this( + serverConfigs.getInteger(AmoroManagementConf.OVERVIEW_CACHE_MAX_SIZE), + serverConfigs.get(AmoroManagementConf.OVERVIEW_CACHE_REFRESH_INTERVAL)); + } + + @VisibleForTesting + public OverviewManager(int maxRecordCount, Duration refreshInterval) { + this.maxRecordCount = maxRecordCount; + ScheduledExecutorService overviewUpdaterScheduler = + Executors.newSingleThreadScheduledExecutor( + new ThreadFactoryBuilder() + .setNameFormat("overview-updater-scheduler-%d") + .setDaemon(true) + .build()); + resetStatusMap(); + + if (refreshInterval.toMillis() > 0) { + overviewUpdaterScheduler.scheduleAtFixedRate( + this::refresh, 1000L, refreshInterval.toMillis(), TimeUnit.MILLISECONDS); + } + } + + public List<OverviewTopTableItem> getAllTopTableItem() { + return ImmutableList.copyOf(allTopTableItem); + } + + public int getTotalCatalog() { + return totalCatalog.get(); + } + + public int getTotalTableCount() { + return totalTableCount.get(); + } + + public long getTotalDataSize() { + return totalDataSize.get(); + } + + public int getTotalCpu() { + return totalCpu.get(); + } + + public long getTotalMemory() { + return totalMemory.get(); + } + + public List<OverviewResourceUsageItem> getResourceUsageHistory(long startTime) { + return resourceUsageHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public List<OverviewDataSizeItem> getDataSizeHistory(long startTime) { + return dataSizeHistory.stream() + .filter(item -> item.getTs() >= startTime) + .collect(Collectors.toList()); + } + + public Map<String, Long> getOptimizingStatus() { + return optimizingStatusCountMap; + } + + @VisibleForTesting + public void refresh() { + long start = System.currentTimeMillis(); + LOG.info("Updating overview cache"); + try { + refreshTableCache(start); + refreshResourceUsage(start); + + } catch (Exception e) { + LOG.error("OverviewRefresher error", e); + } finally { + long end = System.currentTimeMillis(); + LOG.info("Refresher overview cache took {} ms.", end - start); + } + } + + private void refreshTableCache(long ts) { + int totalCatalogs = getAs(CatalogMetaMapper.class, CatalogMetaMapper::selectCatalogCount); + + List<TableRuntimeMeta> metas = + getAs(TableMetaMapper.class, TableMetaMapper::selectTableRuntimeMetas); + AtomicLong totalDataSize = new AtomicLong(); + AtomicInteger totalFileCounts = new AtomicInteger(); + Map<String, OverviewTopTableItem> topTableItemMap = Maps.newHashMap(); + Map<String, Long> optimizingStatusMap = Maps.newHashMap(); + for (TableRuntimeMeta meta : metas) { + Optional<OverviewTopTableItem> optItem = toTopTableItem(meta); + optItem.ifPresent( + tableItem -> { + topTableItemMap.put(tableItem.getTableName(), tableItem); + totalDataSize.addAndGet(tableItem.getTableSize()); + totalFileCounts.addAndGet(tableItem.getFileCount()); + }); + String status = statusToMetricString(meta.getTableStatus()); + if (StringUtils.isNotEmpty(status)) { + optimizingStatusMap.computeIfAbsent(status, ignore -> 0L); Review Comment: ```suggestion optimizingStatusMap.putIfAbsent(status, 0L); ``` ########## amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/OverviewController.java: ########## @@ -38,25 +38,25 @@ /** The controller that handles overview page requests. */ public class OverviewController { - private OverviewCache overviewCache; Review Comment: `OverviewCache ` seems no longer needed, we may remove it. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
