#9905

Committed 23 Aug 2023 06:20AM UTC coverage: 47.785% (-0.1%) from 47.922%

Build # #9905

Build Type

push

travis_ci

Committed by

web-flow

Commit Message

[To rel/1.2][Metric] Fix flush point statistics (#10934)

Run Details

23 of 23 new or added lines in 1 file covered. (100.0%)

79851 of 167106 relevant lines covered (47.78%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterQuotaManager.java

/*
 * 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.iotdb.confignode.manager;

import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq;
import org.apache.iotdb.common.rpc.thrift.TSetThrottleQuotaReq;
import org.apache.iotdb.common.rpc.thrift.TSpaceQuota;
import org.apache.iotdb.common.rpc.thrift.TThrottleQuota;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
import org.apache.iotdb.confignode.client.async.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler;
import org.apache.iotdb.confignode.consensus.request.write.quota.SetSpaceQuotaPlan;
import org.apache.iotdb.confignode.consensus.request.write.quota.SetThrottleQuotaPlan;
import org.apache.iotdb.confignode.manager.partition.PartitionManager;
import org.apache.iotdb.confignode.persistence.quota.QuotaInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowThrottleReq;
import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp;
import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

public class ClusterQuotaManager {

  private static final Logger LOGGER = LoggerFactory.getLogger(ClusterQuotaManager.class);

  private final IManager configManager;
  private final QuotaInfo quotaInfo;
  private final Map<Integer, Long> deviceNum;
  private final Map<Integer, Long> timeSeriesNum;
  private final Map<String, List<Integer>> schemaRegionIdMap;
  private final Map<String, List<Integer>> dataRegionIdMap;
  private final Map<Integer, Long> regionDisk;

  public ClusterQuotaManager(IManager configManager, QuotaInfo quotaInfo) {
    this.configManager = configManager;
    this.quotaInfo = quotaInfo;
    deviceNum = new ConcurrentHashMap<>();
    timeSeriesNum = new ConcurrentHashMap<>();
    schemaRegionIdMap = new HashMap<>();
    dataRegionIdMap = new HashMap<>();
    regionDisk = new ConcurrentHashMap<>();
  }

  public TSStatus setSpaceQuota(TSetSpaceQuotaReq req) {
    if (!checkSpaceQuota(req)) {
      return RpcUtils.getStatus(
          TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
          "The used quota exceeds the preset quota. Please set a larger value.");
    }
    // TODO: Datanode failed to receive rpc
    try {
      TSStatus response =
          configManager
              .getConsensusManager()
              .write(new SetSpaceQuotaPlan(req.getDatabase(), req.getSpaceLimit()));
      if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        Map<Integer, TDataNodeLocation> dataNodeLocationMap =
            configManager.getNodeManager().getRegisteredDataNodeLocations();
        AsyncClientHandler<TSetSpaceQuotaReq, TSStatus> clientHandler =
            new AsyncClientHandler<>(DataNodeRequestType.SET_SPACE_QUOTA, req, dataNodeLocationMap);
        AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);
        return RpcUtils.squashResponseStatusList(clientHandler.getResponseList());
      }
      return response;
    } catch (ConsensusException e) {
      LOGGER.warn(
          String.format(
              "Unexpected error happened while setting space quota on database: %s ",
              req.getDatabase()),
          e);
      // consensus layer related errors
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return res;
    }
  }

  /** If the new quota is smaller than the quota already used, the setting fails. */
  private boolean checkSpaceQuota(TSetSpaceQuotaReq req) {
    for (String database : req.getDatabase()) {
      if (quotaInfo.getSpaceQuotaLimit().containsKey(database)) {
        TSpaceQuota spaceQuota = quotaInfo.getSpaceQuotaUsage().get(database);
        if (req.getSpaceLimit().getDeviceNum() != IoTDBConstant.UNLIMITED_VALUE
            && req.getSpaceLimit().getDeviceNum() != IoTDBConstant.DEFAULT_VALUE
            && spaceQuota.getDeviceNum() > req.getSpaceLimit().getDeviceNum()) {
          return false;
        }
        if (req.getSpaceLimit().getTimeserieNum() != IoTDBConstant.UNLIMITED_VALUE
            && req.getSpaceLimit().getTimeserieNum() != IoTDBConstant.DEFAULT_VALUE
            && spaceQuota.getTimeserieNum() > req.getSpaceLimit().getTimeserieNum()) {
          return false;
        }
        if (req.getSpaceLimit().getDiskSize() != IoTDBConstant.UNLIMITED_VALUE
            && req.getSpaceLimit().getDiskSize() != IoTDBConstant.DEFAULT_VALUE
            && spaceQuota.getDiskSize() > req.getSpaceLimit().getDiskSize()) {
          return false;
        }
      }
    }
    return true;
  }

  public TSpaceQuotaResp showSpaceQuota(List<String> databases) {
    TSpaceQuotaResp showSpaceQuotaResp = new TSpaceQuotaResp();
    if (databases.isEmpty()) {
      showSpaceQuotaResp.setSpaceQuota(quotaInfo.getSpaceQuotaLimit());
      showSpaceQuotaResp.setSpaceQuotaUsage(quotaInfo.getSpaceQuotaUsage());
    } else if (!quotaInfo.getSpaceQuotaLimit().isEmpty()) {
      Map<String, TSpaceQuota> spaceQuotaMap = new HashMap<>();
      Map<String, TSpaceQuota> spaceQuotaUsageMap = new HashMap<>();
      for (String database : databases) {
        if (quotaInfo.getSpaceQuotaLimit().containsKey(database)) {
          spaceQuotaMap.put(database, quotaInfo.getSpaceQuotaLimit().get(database));
          spaceQuotaUsageMap.put(database, quotaInfo.getSpaceQuotaUsage().get(database));
        }
      }
      showSpaceQuotaResp.setSpaceQuota(spaceQuotaMap);
      showSpaceQuotaResp.setSpaceQuotaUsage(spaceQuotaUsageMap);
    }
    showSpaceQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
    return showSpaceQuotaResp;
  }

  public TSpaceQuotaResp getSpaceQuota() {
    TSpaceQuotaResp spaceQuotaResp = new TSpaceQuotaResp();
    if (!quotaInfo.getSpaceQuotaLimit().isEmpty()) {
      spaceQuotaResp.setSpaceQuota(quotaInfo.getSpaceQuotaLimit());
    }
    spaceQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
    return spaceQuotaResp;
  }

  public boolean hasSpaceQuotaLimit() {
    return quotaInfo.getSpaceQuotaLimit().keySet().isEmpty();
  }

  public List<Integer> getSchemaRegionIds() {
    List<Integer> schemaRegionIds = new ArrayList<>();
    getPartitionManager()
        .getSchemaRegionIds(
            new ArrayList<>(quotaInfo.getSpaceQuotaLimit().keySet()), schemaRegionIdMap);
    schemaRegionIdMap.values().forEach(schemaRegionIds::addAll);
    return schemaRegionIds;
  }

  public List<Integer> getDataRegionIds() {
    List<Integer> dataRegionIds = new ArrayList<>();
    getPartitionManager()
        .getDataRegionIds(
            new ArrayList<>(quotaInfo.getSpaceQuotaLimit().keySet()), dataRegionIdMap);
    dataRegionIdMap.values().forEach(dataRegionIds::addAll);
    return dataRegionIds;
  }

  public TSStatus setThrottleQuota(TSetThrottleQuotaReq req) {
    try {
      TSStatus response =
          configManager
              .getConsensusManager()
              .write(new SetThrottleQuotaPlan(req.getUserName(), req.getThrottleQuota()));
      if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        Map<Integer, TDataNodeLocation> dataNodeLocationMap =
            configManager.getNodeManager().getRegisteredDataNodeLocations();
        AsyncClientHandler<TSetThrottleQuotaReq, TSStatus> clientHandler =
            new AsyncClientHandler<>(
                DataNodeRequestType.SET_THROTTLE_QUOTA, req, dataNodeLocationMap);
        AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);
        return RpcUtils.squashResponseStatusList(clientHandler.getResponseList());
      }
      return response;
    } catch (ConsensusException e) {
      LOGGER.warn(
          String.format(
              "Unexpected error happened while setting throttle quota on user: %s ",
              req.getUserName()),
          e);
      // consensus layer related errors
      TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
      res.setMessage(e.getMessage());
      return res;
    }
  }

  public TThrottleQuotaResp showThrottleQuota(TShowThrottleReq req) {
    TThrottleQuotaResp throttleQuotaResp = new TThrottleQuotaResp();
    if (req.getUserName() == null) {
      throttleQuotaResp.setThrottleQuota(quotaInfo.getThrottleQuotaLimit());
    } else {
      Map<String, TThrottleQuota> throttleLimit = new HashMap<>();
      throttleLimit.put(
          req.getUserName(),
          quotaInfo.getThrottleQuotaLimit().get(req.getUserName()) == null
              ? new TThrottleQuota()
              : quotaInfo.getThrottleQuotaLimit().get(req.getUserName()));
      throttleQuotaResp.setThrottleQuota(throttleLimit);
    }
    throttleQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
    return throttleQuotaResp;
  }

  public TThrottleQuotaResp getThrottleQuota() {
    TThrottleQuotaResp throttleQuotaResp = new TThrottleQuotaResp();
    if (!quotaInfo.getThrottleQuotaLimit().isEmpty()) {
      throttleQuotaResp.setThrottleQuota(quotaInfo.getThrottleQuotaLimit());
    }
    throttleQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
    return throttleQuotaResp;
  }

  public Map<String, TSpaceQuota> getSpaceQuotaUsage() {
    return quotaInfo.getSpaceQuotaUsage();
  }

  public Map<Integer, Long> getDeviceNum() {
    return deviceNum;
  }

  public Map<Integer, Long> getTimeSeriesNum() {
    return timeSeriesNum;
  }

  public Map<Integer, Long> getRegionDisk() {
    return regionDisk;
  }

  public void updateSpaceQuotaUsage() {
    AtomicLong deviceCount = new AtomicLong();
    AtomicLong timeSeriesCount = new AtomicLong();
    for (Map.Entry<String, List<Integer>> entry : schemaRegionIdMap.entrySet()) {
      deviceCount.set(0);
      timeSeriesCount.set(0);
      entry
          .getValue()
          .forEach(
              schemaRegionId -> {
                if (deviceNum.containsKey(schemaRegionId)) {
                  deviceCount.addAndGet(deviceCount.get() + deviceNum.get(schemaRegionId));
                }
                if (timeSeriesNum.containsKey(schemaRegionId)) {
                  timeSeriesCount.addAndGet(
                      timeSeriesCount.get() + timeSeriesNum.get(schemaRegionId));
                }
              });
      quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setDeviceNum(deviceCount.get());
      quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setTimeserieNum(timeSeriesCount.get());
    }
    AtomicLong regionDiskCount = new AtomicLong();
    for (Map.Entry<String, List<Integer>> entry : dataRegionIdMap.entrySet()) {
      regionDiskCount.set(0);
      entry
          .getValue()
          .forEach(
              dataRegionId -> {
                if (regionDisk.containsKey(dataRegionId)) {
                  regionDiskCount.addAndGet(regionDisk.get(dataRegionId));
                }
              });
      quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setDiskSize(regionDiskCount.get());
    }
  }

  private PartitionManager getPartitionManager() {
    return configManager.getPartitionManager();
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one
3	* or more contributor license agreements. See the NOTICE file
4	* distributed with this work for additional information
5	* regarding copyright ownership. The ASF licenses this file
6	* to you under the Apache License, Version 2.0 (the
7	* "License"); you may not use this file except in compliance
8	* with the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing,
13	* software distributed under the License is distributed on an
14	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15	* KIND, either express or implied. See the License for the
16	* specific language governing permissions and limitations
17	* under the License.
18	*/
19
20	package org.apache.iotdb.confignode.manager;
21
22	import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
23	import org.apache.iotdb.common.rpc.thrift.TSStatus;
24	import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq;
25	import org.apache.iotdb.common.rpc.thrift.TSetThrottleQuotaReq;
26	import org.apache.iotdb.common.rpc.thrift.TSpaceQuota;
27	import org.apache.iotdb.common.rpc.thrift.TThrottleQuota;
28	import org.apache.iotdb.commons.conf.IoTDBConstant;
29	import org.apache.iotdb.confignode.client.DataNodeRequestType;
30	import org.apache.iotdb.confignode.client.async.AsyncDataNodeClientPool;
31	import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler;
32	import org.apache.iotdb.confignode.consensus.request.write.quota.SetSpaceQuotaPlan;
33	import org.apache.iotdb.confignode.consensus.request.write.quota.SetThrottleQuotaPlan;
34	import org.apache.iotdb.confignode.manager.partition.PartitionManager;
35	import org.apache.iotdb.confignode.persistence.quota.QuotaInfo;
36	import org.apache.iotdb.confignode.rpc.thrift.TShowThrottleReq;
37	import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp;
38	import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp;
39	import org.apache.iotdb.consensus.exception.ConsensusException;
40	import org.apache.iotdb.rpc.RpcUtils;
41	import org.apache.iotdb.rpc.TSStatusCode;
42
43	import org.slf4j.Logger;
44	import org.slf4j.LoggerFactory;
45
46	import java.util.ArrayList;
47	import java.util.HashMap;
48	import java.util.List;
49	import java.util.Map;
50	import java.util.concurrent.ConcurrentHashMap;
51	import java.util.concurrent.atomic.AtomicLong;
52
53	public class ClusterQuotaManager {
54
55	private static final Logger LOGGER = LoggerFactory.getLogger(ClusterQuotaManager.class);	×
56
57	private final IManager configManager;
58	private final QuotaInfo quotaInfo;
59	private final Map<Integer, Long> deviceNum;
60	private final Map<Integer, Long> timeSeriesNum;
61	private final Map<String, List<Integer>> schemaRegionIdMap;
62	private final Map<String, List<Integer>> dataRegionIdMap;
63	private final Map<Integer, Long> regionDisk;
64
65	public ClusterQuotaManager(IManager configManager, QuotaInfo quotaInfo) {	×
66	this.configManager = configManager;	×
67	this.quotaInfo = quotaInfo;	×
68	deviceNum = new ConcurrentHashMap<>();	×
69	timeSeriesNum = new ConcurrentHashMap<>();	×
70	schemaRegionIdMap = new HashMap<>();	×
71	dataRegionIdMap = new HashMap<>();	×
72	regionDisk = new ConcurrentHashMap<>();	×
73	}	×
74
75	public TSStatus setSpaceQuota(TSetSpaceQuotaReq req) {
76	if (!checkSpaceQuota(req)) {	×
77	return RpcUtils.getStatus(	×
78	TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),	×
79	"The used quota exceeds the preset quota. Please set a larger value.");
80	}
81	// TODO: Datanode failed to receive rpc
82	try {
83	TSStatus response =	×
84	configManager
85	.getConsensusManager()	×
86	.write(new SetSpaceQuotaPlan(req.getDatabase(), req.getSpaceLimit()));	×
87	if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {	×
88	Map<Integer, TDataNodeLocation> dataNodeLocationMap =	×
89	configManager.getNodeManager().getRegisteredDataNodeLocations();	×
90	AsyncClientHandler<TSetSpaceQuotaReq, TSStatus> clientHandler =	×
91	new AsyncClientHandler<>(DataNodeRequestType.SET_SPACE_QUOTA, req, dataNodeLocationMap);
92	AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);	×
93	return RpcUtils.squashResponseStatusList(clientHandler.getResponseList());	×
94	}
95	return response;	×
96	} catch (ConsensusException e) {	×
97	LOGGER.warn(	×
98	String.format(	×
99	"Unexpected error happened while setting space quota on database: %s ",
100	req.getDatabase()),	×
101	e);
102	// consensus layer related errors
103	TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());	×
104	res.setMessage(e.getMessage());	×
105	return res;	×
106	}
107	}
108
109	/** If the new quota is smaller than the quota already used, the setting fails. */
110	private boolean checkSpaceQuota(TSetSpaceQuotaReq req) {
111	for (String database : req.getDatabase()) {	×
112	if (quotaInfo.getSpaceQuotaLimit().containsKey(database)) {	×
113	TSpaceQuota spaceQuota = quotaInfo.getSpaceQuotaUsage().get(database);	×
114	if (req.getSpaceLimit().getDeviceNum() != IoTDBConstant.UNLIMITED_VALUE	×
115	&& req.getSpaceLimit().getDeviceNum() != IoTDBConstant.DEFAULT_VALUE	×
116	&& spaceQuota.getDeviceNum() > req.getSpaceLimit().getDeviceNum()) {	×
117	return false;	×
118	}
119	if (req.getSpaceLimit().getTimeserieNum() != IoTDBConstant.UNLIMITED_VALUE	×
120	&& req.getSpaceLimit().getTimeserieNum() != IoTDBConstant.DEFAULT_VALUE	×
121	&& spaceQuota.getTimeserieNum() > req.getSpaceLimit().getTimeserieNum()) {	×
122	return false;	×
123	}
124	if (req.getSpaceLimit().getDiskSize() != IoTDBConstant.UNLIMITED_VALUE	×
125	&& req.getSpaceLimit().getDiskSize() != IoTDBConstant.DEFAULT_VALUE	×
126	&& spaceQuota.getDiskSize() > req.getSpaceLimit().getDiskSize()) {	×
127	return false;	×
128	}
129	}
130	}	×
131	return true;	×
132	}
133
134	public TSpaceQuotaResp showSpaceQuota(List<String> databases) {
135	TSpaceQuotaResp showSpaceQuotaResp = new TSpaceQuotaResp();	×
136	if (databases.isEmpty()) {	×
137	showSpaceQuotaResp.setSpaceQuota(quotaInfo.getSpaceQuotaLimit());	×
138	showSpaceQuotaResp.setSpaceQuotaUsage(quotaInfo.getSpaceQuotaUsage());	×
139	} else if (!quotaInfo.getSpaceQuotaLimit().isEmpty()) {	×
140	Map<String, TSpaceQuota> spaceQuotaMap = new HashMap<>();	×
141	Map<String, TSpaceQuota> spaceQuotaUsageMap = new HashMap<>();	×
142	for (String database : databases) {	×
143	if (quotaInfo.getSpaceQuotaLimit().containsKey(database)) {	×
144	spaceQuotaMap.put(database, quotaInfo.getSpaceQuotaLimit().get(database));	×
145	spaceQuotaUsageMap.put(database, quotaInfo.getSpaceQuotaUsage().get(database));	×
146	}
147	}	×
148	showSpaceQuotaResp.setSpaceQuota(spaceQuotaMap);	×
149	showSpaceQuotaResp.setSpaceQuotaUsage(spaceQuotaUsageMap);	×
150	}
151	showSpaceQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));	×
152	return showSpaceQuotaResp;	×
153	}
154
155	public TSpaceQuotaResp getSpaceQuota() {
156	TSpaceQuotaResp spaceQuotaResp = new TSpaceQuotaResp();	×
157	if (!quotaInfo.getSpaceQuotaLimit().isEmpty()) {	×
158	spaceQuotaResp.setSpaceQuota(quotaInfo.getSpaceQuotaLimit());	×
159	}
160	spaceQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));	×
161	return spaceQuotaResp;	×
162	}
163
164	public boolean hasSpaceQuotaLimit() {
165	return quotaInfo.getSpaceQuotaLimit().keySet().isEmpty();	×
166	}
167
168	public List<Integer> getSchemaRegionIds() {
169	List<Integer> schemaRegionIds = new ArrayList<>();	×
170	getPartitionManager()	×
171	.getSchemaRegionIds(	×
172	new ArrayList<>(quotaInfo.getSpaceQuotaLimit().keySet()), schemaRegionIdMap);	×
173	schemaRegionIdMap.values().forEach(schemaRegionIds::addAll);	×
174	return schemaRegionIds;	×
175	}
176
177	public List<Integer> getDataRegionIds() {
178	List<Integer> dataRegionIds = new ArrayList<>();	×
179	getPartitionManager()	×
180	.getDataRegionIds(	×
181	new ArrayList<>(quotaInfo.getSpaceQuotaLimit().keySet()), dataRegionIdMap);	×
182	dataRegionIdMap.values().forEach(dataRegionIds::addAll);	×
183	return dataRegionIds;	×
184	}
185
186	public TSStatus setThrottleQuota(TSetThrottleQuotaReq req) {
187	try {
188	TSStatus response =	×
189	configManager
190	.getConsensusManager()	×
191	.write(new SetThrottleQuotaPlan(req.getUserName(), req.getThrottleQuota()));	×
192	if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {	×
193	Map<Integer, TDataNodeLocation> dataNodeLocationMap =	×
194	configManager.getNodeManager().getRegisteredDataNodeLocations();	×
195	AsyncClientHandler<TSetThrottleQuotaReq, TSStatus> clientHandler =	×
196	new AsyncClientHandler<>(
197	DataNodeRequestType.SET_THROTTLE_QUOTA, req, dataNodeLocationMap);
198	AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);	×
199	return RpcUtils.squashResponseStatusList(clientHandler.getResponseList());	×
200	}
201	return response;	×
202	} catch (ConsensusException e) {	×
203	LOGGER.warn(	×
204	String.format(	×
205	"Unexpected error happened while setting throttle quota on user: %s ",
206	req.getUserName()),	×
207	e);
208	// consensus layer related errors
209	TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());	×
210	res.setMessage(e.getMessage());	×
211	return res;	×
212	}
213	}
214
215	public TThrottleQuotaResp showThrottleQuota(TShowThrottleReq req) {
216	TThrottleQuotaResp throttleQuotaResp = new TThrottleQuotaResp();	×
217	if (req.getUserName() == null) {	×
218	throttleQuotaResp.setThrottleQuota(quotaInfo.getThrottleQuotaLimit());	×
219	} else {
220	Map<String, TThrottleQuota> throttleLimit = new HashMap<>();	×
221	throttleLimit.put(	×
222	req.getUserName(),	×
223	quotaInfo.getThrottleQuotaLimit().get(req.getUserName()) == null	×
224	? new TThrottleQuota()	×
225	: quotaInfo.getThrottleQuotaLimit().get(req.getUserName()));	×
226	throttleQuotaResp.setThrottleQuota(throttleLimit);	×
227	}
228	throttleQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));	×
229	return throttleQuotaResp;	×
230	}
231
232	public TThrottleQuotaResp getThrottleQuota() {
233	TThrottleQuotaResp throttleQuotaResp = new TThrottleQuotaResp();	×
234	if (!quotaInfo.getThrottleQuotaLimit().isEmpty()) {	×
235	throttleQuotaResp.setThrottleQuota(quotaInfo.getThrottleQuotaLimit());	×
236	}
237	throttleQuotaResp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));	×
238	return throttleQuotaResp;	×
239	}
240
241	public Map<String, TSpaceQuota> getSpaceQuotaUsage() {
242	return quotaInfo.getSpaceQuotaUsage();	×
243	}
244
245	public Map<Integer, Long> getDeviceNum() {
246	return deviceNum;	×
247	}
248
249	public Map<Integer, Long> getTimeSeriesNum() {
250	return timeSeriesNum;	×
251	}
252
253	public Map<Integer, Long> getRegionDisk() {
254	return regionDisk;	×
255	}
256
257	public void updateSpaceQuotaUsage() {
258	AtomicLong deviceCount = new AtomicLong();	×
259	AtomicLong timeSeriesCount = new AtomicLong();	×
260	for (Map.Entry<String, List<Integer>> entry : schemaRegionIdMap.entrySet()) {	×
261	deviceCount.set(0);	×
262	timeSeriesCount.set(0);	×
263	entry	×
264	.getValue()	×
265	.forEach(	×
266	schemaRegionId -> {
267	if (deviceNum.containsKey(schemaRegionId)) {	×
268	deviceCount.addAndGet(deviceCount.get() + deviceNum.get(schemaRegionId));	×
269	}
270	if (timeSeriesNum.containsKey(schemaRegionId)) {	×
271	timeSeriesCount.addAndGet(	×
272	timeSeriesCount.get() + timeSeriesNum.get(schemaRegionId));	×
273	}
274	});	×
275	quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setDeviceNum(deviceCount.get());	×
276	quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setTimeserieNum(timeSeriesCount.get());	×
277	}	×
278	AtomicLong regionDiskCount = new AtomicLong();	×
279	for (Map.Entry<String, List<Integer>> entry : dataRegionIdMap.entrySet()) {	×
280	regionDiskCount.set(0);	×
281	entry	×
282	.getValue()	×
283	.forEach(	×
284	dataRegionId -> {
285	if (regionDisk.containsKey(dataRegionId)) {	×
286	regionDiskCount.addAndGet(regionDisk.get(dataRegionId));	×
287	}
288	});	×
289	quotaInfo.getSpaceQuotaUsage().get(entry.getKey()).setDiskSize(regionDiskCount.get());	×
290	}	×
291	}	×
292
293	private PartitionManager getPartitionManager() {
294	return configManager.getPartitionManager();	×
295	}
296	}

apache / iotdb / #9905

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous