Max Petrov
2 years ago
committed by
GitHub
GitHub
19 changed files with 819 additions and 20 deletions
@ -0,0 +1,19 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.server.common.msg.gateway.metrics; |
||||
|
|
||||
|
public record GatewayMetadata(String connector, long receivedTs, long publishedTs) { |
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.server.transport.mqtt; |
||||
|
|
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
||||
|
|
||||
|
import java.lang.annotation.Retention; |
||||
|
import java.lang.annotation.RetentionPolicy; |
||||
|
|
||||
|
@Retention(RetentionPolicy.RUNTIME) |
||||
|
@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") |
||||
|
public @interface TbMqttTransportComponent { |
||||
|
} |
||||
@ -0,0 +1,113 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.server.transport.mqtt.gateway; |
||||
|
|
||||
|
import jakarta.annotation.PostConstruct; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.transport.TransportService; |
||||
|
import org.thingsboard.server.common.transport.TransportServiceCallback; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.queue.scheduler.SchedulerComponent; |
||||
|
import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; |
||||
|
import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; |
||||
|
import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Service |
||||
|
@TbMqttTransportComponent |
||||
|
public class GatewayMetricsService { |
||||
|
|
||||
|
public static final String GATEWAY_METRICS = "gatewayMetrics"; |
||||
|
|
||||
|
@Value("${transport.mqtt.gateway_metrics_report_interval_sec:60}") |
||||
|
private int metricsReportIntervalSec; |
||||
|
|
||||
|
@Autowired |
||||
|
private SchedulerComponent scheduler; |
||||
|
|
||||
|
@Autowired |
||||
|
private TransportService transportService; |
||||
|
|
||||
|
private Map<DeviceId, GatewayMetricsState> states = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
@PostConstruct |
||||
|
private void init() { |
||||
|
scheduler.scheduleAtFixedRate(this::reportMetrics, metricsReportIntervalSec, metricsReportIntervalSec, TimeUnit.SECONDS); |
||||
|
} |
||||
|
|
||||
|
public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List<GatewayMetadata> data, long serverReceiveTs) { |
||||
|
states.computeIfAbsent(gatewayId, k -> new GatewayMetricsState(sessionInfo)).update(data, serverReceiveTs); |
||||
|
} |
||||
|
|
||||
|
public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId) { |
||||
|
var state = states.get(gatewayId); |
||||
|
if (state != null) { |
||||
|
state.updateSessionInfo(sessionInfo); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void onDeviceDelete(DeviceId deviceId) { |
||||
|
states.remove(deviceId); |
||||
|
} |
||||
|
|
||||
|
public void reportMetrics() { |
||||
|
if (states.isEmpty()) { |
||||
|
return; |
||||
|
} |
||||
|
Map<DeviceId, GatewayMetricsState> statesToReport = states; |
||||
|
states = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
long ts = System.currentTimeMillis(); |
||||
|
|
||||
|
statesToReport.forEach((gatewayId, state) -> { |
||||
|
reportMetrics(state, ts); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void reportMetrics(GatewayMetricsState state, long ts) { |
||||
|
if (state.isEmpty()) { |
||||
|
return; |
||||
|
} |
||||
|
var result = state.getStateResult(); |
||||
|
var kvProto = TransportProtos.KeyValueProto.newBuilder() |
||||
|
.setKey(GATEWAY_METRICS) |
||||
|
.setType(TransportProtos.KeyValueType.JSON_V) |
||||
|
.setJsonV(JacksonUtil.toString(result)) |
||||
|
.build(); |
||||
|
|
||||
|
TransportProtos.TsKvListProto tsKvList = TransportProtos.TsKvListProto.newBuilder() |
||||
|
.setTs(ts) |
||||
|
.addKv(kvProto) |
||||
|
.build(); |
||||
|
|
||||
|
TransportProtos.PostTelemetryMsg telemetryMsg = TransportProtos.PostTelemetryMsg.newBuilder() |
||||
|
.addTsKvList(tsKvList) |
||||
|
.build(); |
||||
|
|
||||
|
transportService.process(state.getSessionInfo(), telemetryMsg, TransportServiceCallback.EMPTY); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,123 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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.thingsboard.server.transport.mqtt.gateway.metrics; |
||||
|
|
||||
|
import lombok.Getter; |
||||
|
import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
|
||||
|
import java.util.HashMap; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||
|
import java.util.concurrent.locks.Lock; |
||||
|
import java.util.concurrent.locks.ReentrantLock; |
||||
|
|
||||
|
public class GatewayMetricsState { |
||||
|
|
||||
|
private final Map<String, ConnectorMetricsState> connectors; |
||||
|
private final Lock updateLock; |
||||
|
|
||||
|
@Getter |
||||
|
private volatile TransportProtos.SessionInfoProto sessionInfo; |
||||
|
|
||||
|
public GatewayMetricsState(TransportProtos.SessionInfoProto sessionInfo) { |
||||
|
this.connectors = new HashMap<>(); |
||||
|
this.updateLock = new ReentrantLock(); |
||||
|
this.sessionInfo = sessionInfo; |
||||
|
} |
||||
|
|
||||
|
public void updateSessionInfo(TransportProtos.SessionInfoProto sessionInfo) { |
||||
|
this.sessionInfo = sessionInfo; |
||||
|
} |
||||
|
|
||||
|
public void update(List<GatewayMetadata> metricsData, long serverReceiveTs) { |
||||
|
updateLock.lock(); |
||||
|
try { |
||||
|
metricsData.forEach(data -> { |
||||
|
connectors.computeIfAbsent(data.connector(), k -> new ConnectorMetricsState()).update(data, serverReceiveTs); |
||||
|
}); |
||||
|
} finally { |
||||
|
updateLock.unlock(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public Map<String, ConnectorMetricsResult> getStateResult() { |
||||
|
Map<String, ConnectorMetricsResult> result = new HashMap<>(); |
||||
|
updateLock.lock(); |
||||
|
try { |
||||
|
connectors.forEach((name, state) -> result.put(name, state.getResult())); |
||||
|
connectors.clear(); |
||||
|
} finally { |
||||
|
updateLock.unlock(); |
||||
|
} |
||||
|
|
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
public boolean isEmpty() { |
||||
|
return connectors.isEmpty(); |
||||
|
} |
||||
|
|
||||
|
private static class ConnectorMetricsState { |
||||
|
private final AtomicInteger count; |
||||
|
private final AtomicLong gwLatencySum; |
||||
|
private final AtomicLong transportLatencySum; |
||||
|
private volatile long minGwLatency; |
||||
|
private volatile long maxGwLatency; |
||||
|
private volatile long minTransportLatency; |
||||
|
private volatile long maxTransportLatency; |
||||
|
|
||||
|
private ConnectorMetricsState() { |
||||
|
this.count = new AtomicInteger(0); |
||||
|
this.gwLatencySum = new AtomicLong(0); |
||||
|
this.transportLatencySum = new AtomicLong(0); |
||||
|
} |
||||
|
|
||||
|
private void update(GatewayMetadata metricsData, long serverReceiveTs) { |
||||
|
long gwLatency = metricsData.publishedTs() - metricsData.receivedTs(); |
||||
|
long transportLatency = serverReceiveTs - metricsData.publishedTs(); |
||||
|
count.incrementAndGet(); |
||||
|
gwLatencySum.addAndGet(gwLatency); |
||||
|
transportLatencySum.addAndGet(transportLatency); |
||||
|
if (minGwLatency == 0 || minGwLatency > gwLatency) { |
||||
|
minGwLatency = gwLatency; |
||||
|
} |
||||
|
if (maxGwLatency < gwLatency) { |
||||
|
maxGwLatency = gwLatency; |
||||
|
} |
||||
|
if (minTransportLatency == 0 || minTransportLatency > transportLatency) { |
||||
|
minTransportLatency = transportLatency; |
||||
|
} |
||||
|
if (maxTransportLatency < transportLatency) { |
||||
|
maxTransportLatency = transportLatency; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private ConnectorMetricsResult getResult() { |
||||
|
long count = this.count.get(); |
||||
|
long avgGwLatency = gwLatencySum.get() / count; |
||||
|
long avgTransportLatency = transportLatencySum.get() / count; |
||||
|
return new ConnectorMetricsResult(avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public record ConnectorMetricsResult(long avgGwLatency, long minGwLatency, long maxGwLatency, |
||||
|
long avgTransportLatency, long minTransportLatency, long maxTransportLatency) { |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue