mp-loki
8 years ago
41 changed files with 774 additions and 115 deletions
File diff suppressed because one or more lines are too long
@ -0,0 +1,90 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.config; |
|||
|
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.security.core.Authentication; |
|||
import org.springframework.security.core.context.SecurityContextHolder; |
|||
import org.springframework.stereotype.Component; |
|||
import org.springframework.web.filter.GenericFilterBean; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.tools.TbRateLimits; |
|||
import org.thingsboard.server.common.msg.tools.TbRateLimitsException; |
|||
import org.thingsboard.server.exception.ThingsboardErrorResponseHandler; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
|
|||
import javax.servlet.FilterChain; |
|||
import javax.servlet.ServletException; |
|||
import javax.servlet.ServletRequest; |
|||
import javax.servlet.ServletResponse; |
|||
import javax.servlet.http.HttpServletResponse; |
|||
import java.io.IOException; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
|
|||
@Component |
|||
public class RateLimitProcessingFilter extends GenericFilterBean { |
|||
|
|||
@Value("${server.rest.limits.tenant.enabled:false}") |
|||
private boolean perTenantLimitsEnabled; |
|||
@Value("${server.rest.limits.tenant.configuration:}") |
|||
private String perTenantLimitsConfiguration; |
|||
@Value("${server.rest.limits.customer.enabled:false}") |
|||
private boolean perCustomerLimitsEnabled; |
|||
@Value("${server.rest.limits.customer.configuration:}") |
|||
private String perCustomerLimitsConfiguration; |
|||
|
|||
@Autowired |
|||
private ThingsboardErrorResponseHandler errorResponseHandler; |
|||
|
|||
private ConcurrentMap<TenantId, TbRateLimits> perTenantLimits = new ConcurrentHashMap<>(); |
|||
private ConcurrentMap<CustomerId, TbRateLimits> perCustomerLimits = new ConcurrentHashMap<>(); |
|||
|
|||
@Override |
|||
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException { |
|||
SecurityUser user = getCurrentUser(); |
|||
if (user != null && !user.isSystemAdmin()) { |
|||
if (perTenantLimitsEnabled) { |
|||
TbRateLimits rateLimits = perTenantLimits.computeIfAbsent(user.getTenantId(), id -> new TbRateLimits(perTenantLimitsConfiguration)); |
|||
if (!rateLimits.tryConsume()) { |
|||
errorResponseHandler.handle(new TbRateLimitsException(EntityType.TENANT), (HttpServletResponse) response); |
|||
return; |
|||
} |
|||
} |
|||
if (perCustomerLimitsEnabled && user.isCustomerUser()) { |
|||
TbRateLimits rateLimits = perCustomerLimits.computeIfAbsent(user.getCustomerId(), id -> new TbRateLimits(perCustomerLimitsConfiguration)); |
|||
if (!rateLimits.tryConsume()) { |
|||
errorResponseHandler.handle(new TbRateLimitsException(EntityType.CUSTOMER), (HttpServletResponse) response); |
|||
return; |
|||
} |
|||
} |
|||
} |
|||
chain.doFilter(request, response); |
|||
} |
|||
|
|||
protected SecurityUser getCurrentUser() { |
|||
Authentication authentication = SecurityContextHolder.getContext().getAuthentication(); |
|||
if (authentication != null && authentication.getPrincipal() instanceof SecurityUser) { |
|||
return (SecurityUser) authentication.getPrincipal(); |
|||
} else { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,101 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.rule.engine.action; |
|||
|
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import com.google.gson.Gson; |
|||
import com.google.gson.JsonObject; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.rule.engine.api.*; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgDataType; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
import java.util.UUID; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicLong; |
|||
|
|||
import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; |
|||
|
|||
@Slf4j |
|||
@RuleNode( |
|||
type = ComponentType.ACTION, |
|||
name = "message count", |
|||
configClazz = TbMsgCountNodeConfiguration.class, |
|||
nodeDescription = "Count incoming messages", |
|||
nodeDetails = "Count incoming messages for specified interval and produces POST_TELEMETRY_REQUEST msg with messages count", |
|||
icon = "functions", |
|||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|||
configDirective = "tbActionNodeMsgCountConfig" |
|||
) |
|||
public class TbMsgCountNode implements TbNode { |
|||
|
|||
private static final String TB_MSG_COUNT_NODE_MSG = "TbMsgCountNodeMsg"; |
|||
|
|||
private AtomicLong messagesProcessed = new AtomicLong(0); |
|||
private final Gson gson = new Gson(); |
|||
private UUID nextTickId; |
|||
private long delay; |
|||
private String telemetryPrefix; |
|||
private long lastScheduledTs; |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
TbMsgCountNodeConfiguration config = TbNodeUtils.convert(configuration, TbMsgCountNodeConfiguration.class); |
|||
this.delay = TimeUnit.SECONDS.toMillis(config.getInterval()); |
|||
this.telemetryPrefix = config.getTelemetryPrefix(); |
|||
scheduleTickMsg(ctx); |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) { |
|||
if (msg.getType().equals(TB_MSG_COUNT_NODE_MSG) && msg.getId().equals(nextTickId)) { |
|||
JsonObject telemetryJson = new JsonObject(); |
|||
telemetryJson.addProperty(this.telemetryPrefix + "_" + ctx.getNodeId(), messagesProcessed.longValue()); |
|||
|
|||
messagesProcessed = new AtomicLong(0); |
|||
|
|||
TbMsgMetaData metaData = new TbMsgMetaData(); |
|||
metaData.putValue("delta", Long.toString(System.currentTimeMillis() - lastScheduledTs + delay)); |
|||
|
|||
TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), metaData, TbMsgDataType.JSON, gson.toJson(telemetryJson), null, null, 0L); |
|||
ctx.tellNext(tbMsg, SUCCESS); |
|||
scheduleTickMsg(ctx); |
|||
} else { |
|||
messagesProcessed.incrementAndGet(); |
|||
} |
|||
} |
|||
|
|||
private void scheduleTickMsg(TbContext ctx) { |
|||
long curTs = System.currentTimeMillis(); |
|||
if (lastScheduledTs == 0L) { |
|||
lastScheduledTs = curTs; |
|||
} |
|||
lastScheduledTs = lastScheduledTs + delay; |
|||
long curDelay = Math.max(0L, (lastScheduledTs - curTs)); |
|||
TbMsg tickMsg = ctx.newMsg(TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); |
|||
nextTickId = tickMsg.getId(); |
|||
ctx.tellSelf(tickMsg, curDelay); |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
} |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.rule.engine.action; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.rule.engine.api.NodeConfiguration; |
|||
|
|||
@Data |
|||
public class TbMsgCountNodeConfiguration implements NodeConfiguration<TbMsgCountNodeConfiguration> { |
|||
|
|||
private String telemetryPrefix; |
|||
private int interval; |
|||
|
|||
@Override |
|||
public TbMsgCountNodeConfiguration defaultConfiguration() { |
|||
TbMsgCountNodeConfiguration configuration = new TbMsgCountNodeConfiguration(); |
|||
configuration.setInterval(1); |
|||
configuration.setTelemetryPrefix("messageCount"); |
|||
return configuration; |
|||
} |
|||
} |
|||
File diff suppressed because one or more lines are too long
Loading…
Reference in new issue