|
|
@ -15,12 +15,11 @@ |
|
|
*/ |
|
|
*/ |
|
|
package org.thingsboard.script.api; |
|
|
package org.thingsboard.script.api; |
|
|
|
|
|
|
|
|
|
|
|
import com.google.common.util.concurrent.FutureCallback; |
|
|
import com.google.common.util.concurrent.Futures; |
|
|
import com.google.common.util.concurrent.Futures; |
|
|
import com.google.common.util.concurrent.ListenableFuture; |
|
|
import com.google.common.util.concurrent.ListenableFuture; |
|
|
import com.google.common.util.concurrent.MoreExecutors; |
|
|
import com.google.common.util.concurrent.MoreExecutors; |
|
|
import lombok.Getter; |
|
|
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
|
|
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|
|
import org.thingsboard.server.common.data.ApiUsageRecordKey; |
|
|
import org.thingsboard.server.common.data.ApiUsageRecordKey; |
|
|
import org.thingsboard.server.common.data.id.CustomerId; |
|
|
import org.thingsboard.server.common.data.id.CustomerId; |
|
|
@ -32,42 +31,70 @@ import java.util.Map; |
|
|
import java.util.Optional; |
|
|
import java.util.Optional; |
|
|
import java.util.UUID; |
|
|
import java.util.UUID; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
|
|
|
import java.util.concurrent.Executor; |
|
|
import java.util.concurrent.Executors; |
|
|
import java.util.concurrent.Executors; |
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
|
|
|
import java.util.concurrent.TimeUnit; |
|
|
import java.util.concurrent.atomic.AtomicInteger; |
|
|
import java.util.concurrent.atomic.AtomicInteger; |
|
|
|
|
|
|
|
|
import static java.lang.String.format; |
|
|
import static java.lang.String.format; |
|
|
|
|
|
|
|
|
/** |
|
|
|
|
|
* Created by ashvayka on 26.09.18. |
|
|
|
|
|
*/ |
|
|
|
|
|
@Slf4j |
|
|
@Slf4j |
|
|
public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
public abstract class AbstractScriptInvokeService implements ScriptInvokeService { |
|
|
|
|
|
|
|
|
|
|
|
protected Map<UUID, DisableListInfo> disabledScripts = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
|
private final Optional<TbApiUsageStateClient> apiUsageStateClient; |
|
|
private final Optional<TbApiUsageStateClient> apiUsageStateClient; |
|
|
private final Optional<TbApiUsageReportClient> apiUsageReportClient; |
|
|
private final Optional<TbApiUsageReportClient> apiUsageReportClient; |
|
|
|
|
|
private final AtomicInteger pushedMsgs = new AtomicInteger(0); |
|
|
|
|
|
private final AtomicInteger invokeMsgs = new AtomicInteger(0); |
|
|
|
|
|
private final AtomicInteger evalMsgs = new AtomicInteger(0); |
|
|
|
|
|
protected final AtomicInteger failedMsgs = new AtomicInteger(0); |
|
|
|
|
|
protected final AtomicInteger timeoutMsgs = new AtomicInteger(0); |
|
|
|
|
|
|
|
|
|
|
|
private final FutureCallback<UUID> evalCallback = new ScriptStatCallback<>(evalMsgs, timeoutMsgs, failedMsgs); |
|
|
|
|
|
private final FutureCallback<Object> invokeCallback = new ScriptStatCallback<>(invokeMsgs, timeoutMsgs, failedMsgs); |
|
|
|
|
|
|
|
|
protected ScheduledExecutorService timeoutExecutorService; |
|
|
protected ScheduledExecutorService timeoutExecutorService; |
|
|
protected Map<UUID, String> scriptIdToNameMap = new ConcurrentHashMap<>(); |
|
|
|
|
|
protected Map<UUID, DisableListInfo> disabledFunctions = new ConcurrentHashMap<>(); |
|
|
protected AbstractScriptInvokeService(Optional<TbApiUsageStateClient> apiUsageStateClient, Optional<TbApiUsageReportClient> apiUsageReportClient) { |
|
|
|
|
|
|
|
|
@Getter |
|
|
|
|
|
@Value("${js.max_total_args_size:100000}") |
|
|
|
|
|
private long maxTotalArgsSize; |
|
|
|
|
|
@Getter |
|
|
|
|
|
@Value("${js.max_result_size:300000}") |
|
|
|
|
|
private long maxResultSize; |
|
|
|
|
|
@Getter |
|
|
|
|
|
@Value("${js.max_script_body_size:50000}") |
|
|
|
|
|
private long maxScriptBodySize; |
|
|
|
|
|
|
|
|
|
|
|
protected AbstractJsInvokeService(Optional<TbApiUsageStateClient> apiUsageStateClient, Optional<TbApiUsageReportClient> apiUsageReportClient) { |
|
|
|
|
|
this.apiUsageStateClient = apiUsageStateClient; |
|
|
this.apiUsageStateClient = apiUsageStateClient; |
|
|
this.apiUsageReportClient = apiUsageReportClient; |
|
|
this.apiUsageReportClient = apiUsageReportClient; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public void init(long maxRequestsTimeout) { |
|
|
protected long getMaxEvalRequestsTimeout() { |
|
|
if (maxRequestsTimeout > 0) { |
|
|
return getMaxInvokeRequestsTimeout(); |
|
|
timeoutExecutorService = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("nashorn-js-timeout")); |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
protected abstract long getMaxInvokeRequestsTimeout(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract long getMaxScriptBodySize(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract long getMaxTotalArgsSize(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract long getMaxResultSize(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract int getMaxBlackListDurationSec(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract int getMaxErrors(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract boolean isStatsEnabled(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract String getStatsName(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract Executor getCallbackExecutor(); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract boolean isScriptPresent(UUID scriptId); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract ListenableFuture<UUID> doEvalScript(ScriptType scriptType, String scriptBody, UUID scriptId, String[] argNames); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract ListenableFuture<Object> doInvokeFunction(UUID scriptId, Object[] args); |
|
|
|
|
|
|
|
|
|
|
|
protected abstract void doRelease(UUID scriptId) throws Exception; |
|
|
|
|
|
|
|
|
|
|
|
public void init() { |
|
|
|
|
|
if (getMaxEvalRequestsTimeout() > 0 || getMaxInvokeRequestsTimeout() > 0) { |
|
|
|
|
|
timeoutExecutorService = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("script-timeout")); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -77,26 +104,39 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
public void printStats() { |
|
|
|
|
|
if (isStatsEnabled()) { |
|
|
|
|
|
int pushed = pushedMsgs.getAndSet(0); |
|
|
|
|
|
int invoked = invokeMsgs.getAndSet(0); |
|
|
|
|
|
int evaluated = evalMsgs.getAndSet(0); |
|
|
|
|
|
int failed = failedMsgs.getAndSet(0); |
|
|
|
|
|
int timedOut = timeoutMsgs.getAndSet(0); |
|
|
|
|
|
if (pushed > 0 || invoked > 0 || evaluated > 0 || failed > 0 || timedOut > 0) { |
|
|
|
|
|
log.info("{}: pushed [{}] received [{}] invoke [{}] eval [{}] failed [{}] timedOut [{}]", |
|
|
|
|
|
getStatsName(), pushed, invoked + evaluated, invoked, evaluated, failed, timedOut); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public ListenableFuture<UUID> eval(TenantId tenantId, JsScriptType scriptType, String scriptBody, String... argNames) { |
|
|
public ListenableFuture<UUID> eval(TenantId tenantId, ScriptType scriptType, String scriptBody, String... argNames) { |
|
|
if (!apiUsageStateClient.isPresent() || apiUsageStateClient.get().getApiUsageState(tenantId).isJsExecEnabled()) { |
|
|
if (!apiUsageStateClient.isPresent() || apiUsageStateClient.get().getApiUsageState(tenantId).isJsExecEnabled()) { |
|
|
if (scriptBodySizeExceeded(scriptBody)) { |
|
|
if (scriptBodySizeExceeded(scriptBody)) { |
|
|
return error(format("Script body exceeds maximum allowed size of %s symbols", getMaxScriptBodySize())); |
|
|
return error(format("Script body exceeds maximum allowed size of %s symbols", getMaxScriptBodySize())); |
|
|
} |
|
|
} |
|
|
UUID scriptId = UUID.randomUUID(); |
|
|
UUID scriptId = UUID.randomUUID(); |
|
|
String functionName = "invokeInternal_" + scriptId.toString().replace('-', '_'); |
|
|
pushedMsgs.incrementAndGet(); |
|
|
String jsScript = generateJsScript(scriptType, functionName, scriptBody, argNames); |
|
|
return withTimeoutAndStatsCallback(doEvalScript(scriptType, scriptBody, scriptId, argNames), evalCallback, getMaxEvalRequestsTimeout()); |
|
|
return doEval(scriptId, functionName, jsScript); |
|
|
|
|
|
} else { |
|
|
} else { |
|
|
return error("JS Execution is disabled due to API limits!"); |
|
|
return error("Script Execution is disabled due to API limits!"); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public ListenableFuture<String> invokeFunction(TenantId tenantId, CustomerId customerId, UUID scriptId, Object... args) { |
|
|
public ListenableFuture<String> invokeScript(TenantId tenantId, CustomerId customerId, UUID scriptId, Object... args) { |
|
|
if (!apiUsageStateClient.isPresent() || apiUsageStateClient.get().getApiUsageState(tenantId).isJsExecEnabled()) { |
|
|
if (!apiUsageStateClient.isPresent() || apiUsageStateClient.get().getApiUsageState(tenantId).isJsExecEnabled()) { |
|
|
String functionName = scriptIdToNameMap.get(scriptId); |
|
|
if (!isScriptPresent(scriptId)) { |
|
|
if (functionName == null) { |
|
|
|
|
|
return error("No compiled script found for scriptId: [" + scriptId + "]!"); |
|
|
return error("No compiled script found for scriptId: [" + scriptId + "]!"); |
|
|
} |
|
|
} |
|
|
if (!isDisabled(scriptId)) { |
|
|
if (!isDisabled(scriptId)) { |
|
|
@ -104,13 +144,17 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
return scriptExecutionError(scriptId, format("Script input arguments exceed maximum allowed total args size of %s symbols", getMaxTotalArgsSize())); |
|
|
return scriptExecutionError(scriptId, format("Script input arguments exceed maximum allowed total args size of %s symbols", getMaxTotalArgsSize())); |
|
|
} |
|
|
} |
|
|
apiUsageReportClient.ifPresent(client -> client.report(tenantId, customerId, ApiUsageRecordKey.JS_EXEC_COUNT, 1)); |
|
|
apiUsageReportClient.ifPresent(client -> client.report(tenantId, customerId, ApiUsageRecordKey.JS_EXEC_COUNT, 1)); |
|
|
return Futures.transformAsync(doInvokeFunction(scriptId, functionName, args), output -> { |
|
|
pushedMsgs.incrementAndGet(); |
|
|
|
|
|
log.trace("invokeScript uuid {} with timeout {}ms", scriptId, getMaxInvokeRequestsTimeout()); |
|
|
|
|
|
var resultFuture = Futures.transformAsync(doInvokeFunction(scriptId, args), output -> { |
|
|
String result = output.toString(); |
|
|
String result = output.toString(); |
|
|
if (resultSizeExceeded(result)) { |
|
|
if (resultSizeExceeded(result)) { |
|
|
return scriptExecutionError(scriptId, format("Script invocation result exceeds maximum allowed size of %s symbols", getMaxResultSize())); |
|
|
return scriptExecutionError(scriptId, format("Script invocation result exceeds maximum allowed size of %s symbols", getMaxResultSize())); |
|
|
} |
|
|
} |
|
|
return Futures.immediateFuture(result); |
|
|
return Futures.immediateFuture(result); |
|
|
}, MoreExecutors.directExecutor()); |
|
|
}, MoreExecutors.directExecutor()); |
|
|
|
|
|
|
|
|
|
|
|
return withTimeoutAndStatsCallback(resultFuture, invokeCallback, getMaxInvokeRequestsTimeout()); |
|
|
} else { |
|
|
} else { |
|
|
String message = "Script invocation is blocked due to maximum error count " |
|
|
String message = "Script invocation is blocked due to maximum error count " |
|
|
+ getMaxErrors() + ", scriptId " + scriptId + "!"; |
|
|
+ getMaxErrors() + ", scriptId " + scriptId + "!"; |
|
|
@ -122,14 +166,20 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private <T extends V, V> ListenableFuture<T> withTimeoutAndStatsCallback(ListenableFuture<T> future, FutureCallback<V> statsCallback, long timeout) { |
|
|
|
|
|
if (timeout > 0) { |
|
|
|
|
|
future = Futures.withTimeout(future, timeout, TimeUnit.MILLISECONDS, timeoutExecutorService); |
|
|
|
|
|
} |
|
|
|
|
|
Futures.addCallback(future, statsCallback, getCallbackExecutor()); |
|
|
|
|
|
return future; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public ListenableFuture<Void> release(UUID scriptId) { |
|
|
public ListenableFuture<Void> release(UUID scriptId) { |
|
|
String functionName = scriptIdToNameMap.get(scriptId); |
|
|
if (isScriptPresent(scriptId)) { |
|
|
if (functionName != null) { |
|
|
|
|
|
try { |
|
|
try { |
|
|
scriptIdToNameMap.remove(scriptId); |
|
|
disabledScripts.remove(scriptId); |
|
|
disabledFunctions.remove(scriptId); |
|
|
doRelease(scriptId); |
|
|
doRelease(scriptId, functionName); |
|
|
|
|
|
} catch (Exception e) { |
|
|
} catch (Exception e) { |
|
|
return Futures.immediateFailedFuture(e); |
|
|
return Futures.immediateFailedFuture(e); |
|
|
} |
|
|
} |
|
|
@ -137,21 +187,18 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
return Futures.immediateFuture(null); |
|
|
return Futures.immediateFuture(null); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
protected abstract ListenableFuture<UUID> doEval(UUID scriptId, String functionName, String scriptBody); |
|
|
private boolean isDisabled(UUID scriptId) { |
|
|
|
|
|
DisableListInfo errorCount = disabledScripts.get(scriptId); |
|
|
protected abstract ListenableFuture<Object> doInvokeFunction(UUID scriptId, String functionName, Object[] args); |
|
|
if (errorCount != null) { |
|
|
|
|
|
if (errorCount.getExpirationTime() <= System.currentTimeMillis()) { |
|
|
protected abstract void doRelease(UUID scriptId, String functionName) throws Exception; |
|
|
disabledScripts.remove(scriptId); |
|
|
|
|
|
return false; |
|
|
protected abstract int getMaxErrors(); |
|
|
} else { |
|
|
|
|
|
return errorCount.get() >= getMaxErrors(); |
|
|
protected abstract long getMaxBlacklistDuration(); |
|
|
} |
|
|
|
|
|
} else { |
|
|
protected void onScriptExecutionError(UUID scriptId, Throwable t, String scriptBody) { |
|
|
return false; |
|
|
DisableListInfo disableListInfo = disabledFunctions.computeIfAbsent(scriptId, key -> new DisableListInfo()); |
|
|
} |
|
|
log.warn("Script has exception and will increment counter {} on disabledFunctions for id {}, exception {}, cause {}, scriptBody {}", |
|
|
|
|
|
disableListInfo.get(), scriptId, t, t.getCause(), scriptBody); |
|
|
|
|
|
disableListInfo.incrementAndGet(); |
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private boolean scriptBodySizeExceeded(String scriptBody) { |
|
|
private boolean scriptBodySizeExceeded(String scriptBody) { |
|
|
@ -175,31 +222,17 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
return result.length() > getMaxResultSize(); |
|
|
return result.length() > getMaxResultSize(); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private String generateJsScript(JsScriptType scriptType, String functionName, String scriptBody, String... argNames) { |
|
|
|
|
|
if (scriptType == JsScriptType.RULE_NODE_SCRIPT) { |
|
|
|
|
|
return RuleNodeScriptFactory.generateRuleNodeScript(functionName, scriptBody, argNames); |
|
|
|
|
|
} |
|
|
|
|
|
throw new RuntimeException("No script factory implemented for scriptType: " + scriptType); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private boolean isDisabled(UUID scriptId) { |
|
|
|
|
|
DisableListInfo errorCount = disabledFunctions.get(scriptId); |
|
|
|
|
|
if (errorCount != null) { |
|
|
|
|
|
if (errorCount.getExpirationTime() <= System.currentTimeMillis()) { |
|
|
|
|
|
disabledFunctions.remove(scriptId); |
|
|
|
|
|
return false; |
|
|
|
|
|
} else { |
|
|
|
|
|
return errorCount.get() >= getMaxErrors(); |
|
|
|
|
|
} |
|
|
|
|
|
} else { |
|
|
|
|
|
return false; |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private <T> ListenableFuture<T> error(String message) { |
|
|
private <T> ListenableFuture<T> error(String message) { |
|
|
return Futures.immediateFailedFuture(new RuntimeException(message)); |
|
|
return Futures.immediateFailedFuture(new RuntimeException(message)); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
protected void onScriptExecutionError(UUID scriptId, Throwable t, String scriptBody) { |
|
|
|
|
|
DisableListInfo disableListInfo = disabledScripts.computeIfAbsent(scriptId, key -> new DisableListInfo()); |
|
|
|
|
|
log.warn("Script has exception and will increment counter {} on disabledFunctions for id {}, exception {}, cause {}, scriptBody {}", |
|
|
|
|
|
disableListInfo.get(), scriptId, t, t.getCause(), scriptBody); |
|
|
|
|
|
disableListInfo.incrementAndGet(); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
private <T> ListenableFuture<T> scriptExecutionError(UUID scriptId, String errorMsg) { |
|
|
private <T> ListenableFuture<T> scriptExecutionError(UUID scriptId, String errorMsg) { |
|
|
RuntimeException error = new RuntimeException(errorMsg); |
|
|
RuntimeException error = new RuntimeException(errorMsg); |
|
|
onScriptExecutionError(scriptId, error, null); |
|
|
onScriptExecutionError(scriptId, error, null); |
|
|
@ -220,7 +253,7 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
|
|
|
|
|
|
public int incrementAndGet() { |
|
|
public int incrementAndGet() { |
|
|
int result = counter.incrementAndGet(); |
|
|
int result = counter.incrementAndGet(); |
|
|
expirationTime = System.currentTimeMillis() + getMaxBlacklistDuration(); |
|
|
expirationTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(getMaxBlackListDurationSec()); |
|
|
return result; |
|
|
return result; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -228,4 +261,5 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { |
|
|
return expirationTime; |
|
|
return expirationTime; |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
} |
|
|
} |
Andrii Shvaika
4 years ago