Java 混沌实验执行器 chaosblade-exec-jvm

Posted i++;

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java 混沌实验执行器 chaosblade-exec-jvm相关的知识,希望对你有一定的参考价值。

关于混沌实验的意义和作用这里就不说了,这里只说实现。
Java项目混沌实验整体实现思路就是通过java agent 和字节码增强技术注入异常。
针对 Java 注入的异常有:
  1. 线程池被占用
  2. 连接池被占用
  3. 方法延迟
  4. 抛异常
  5. 指定返回值
  6. CPU 满载
  7. 内存溢出
  8. CodeCacheFilling
下面依次展开
 @Override
 public void full(final ThreadPoolExecutor threadPoolExecutor) {
        if (threadPoolExecutor == null) {
            LOGGER.warn("threadPoolExecutor is null");
            return;
        }
        if (futureCache.size() > 0) {
            LOGGER.info("thread pool has started");
            return;
        }
        isRunning = true;
        final int activeCount = threadPoolExecutor.getActiveCount();
        final int corePoolSize = threadPoolExecutor.getCorePoolSize();
        final int maximumPoolSize = threadPoolExecutor.getMaximumPoolSize();
        LOGGER.info("start execute thread pool full, activeCount: {}, corePoolSize: {}, maximumPoolSize: {}",
            activeCount, corePoolSize, maximumPoolSize);

        synchronized (lock) {
            if (executorService == null || executorService.isShutdown() || executorService.isTerminated()) {
                executorService = ThreadUtil.createScheduledExecutorService();
            }
        }
        executorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    for (int i = 0; i < maximumPoolSize; i++) {
                        // 不断的提交线程, 这个线程里面就是一个sleep, 所以会霸占这线程
                        Future<?> future = threadPoolExecutor.submit(new InterruptableRunnable());
                        if (future != null) {
                            // 放入到一个数组中,等恢复的时候,再把数组中的线程关闭掉
                            futureCache.add(future);
                        }
                    }
                } catch (RejectedExecutionException e) {
                    LOGGER.info("has triggered thread pool full");
                } catch (Exception e) {
                    LOGGER.error("execute thread pool full exception", e);
                }
            }
        }, 0, 10, TimeUnit.SECONDS);
}
@Override
public void full(final DataSource dataSource) {
        if (dataSource == null) {
            LOGGER.warn("dataSource is null");
            return;
        }
        if (connectionHolder.size() > 0) {
            LOGGER.info("connection pool full has started");
            return;
        }
        isRunning = true;

        int maxPoolSize = getMaxPoolSize();
        final int poolSize = maxPoolSize <= 0 ? DEFAULT_MAX_POOL_SIZE : maxPoolSize;
        LOGGER.info("Start execute connection pool full, poolSize: {}", poolSize);

        synchronized (lock) {
            if (executorService == null || executorService.isShutdown() || executorService.isTerminated()) {
                executorService = ThreadUtil.createScheduledExecutorService();
            }
        }
        executorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    for (int i = 0; i < poolSize; i++) {
                        // 这里持续获取连接,但不释放
                        Connection connection = dataSource.getConnection();
                        if (connection != null) {
                            // 将连接放到一个数组中,在恢复的时候, 关闭掉
                            connectionHolder.add(connection);
                        }
                    }
                    LOGGER.info("execute connection pool full success");
                } catch (SQLException e) {
                    LOGGER.info("connection pool full, {}", e.getMessage());
                } catch (Exception e) {
                    LOGGER.warn("get database connection exception", e);
                }
            }
        }, 0, 10, TimeUnit.SECONDS);
}
@Override
public void run(EnhancerModel enhancerModel) throws Exception {
        // 要延迟的时间
        String time = enhancerModel.getActionFlag(timeFlagSpec.getName());
        Integer sleepTimeInMillis = Integer.valueOf(time);
        int offset = 0;
        String offsetTime = enhancerModel.getActionFlag(timeOffsetFlagSpec.getName());
        if (!StringUtil.isBlank(offsetTime)) {
            offset = Integer.valueOf(offsetTime);
        }
        TimeoutExecutor timeoutExecutor = enhancerModel.getTimeoutExecutor();
        if (timeoutExecutor != null) {
            long timeoutInMillis = timeoutExecutor.getTimeoutInMillis();
            if (timeoutInMillis > 0 && timeoutInMillis < sleepTimeInMillis) {
                sleep(timeoutInMillis, 0);
                timeoutExecutor.run(enhancerModel);
                return;
            }
        }
        sleep(sleepTimeInMillis, offset);
}

@Override
public void sleep(long timeInMillis, int offsetInMillis) {
Random random = new Random();
int offset = 0;
if (offsetInMillis > 0) {
offset = random.nextInt(offsetInMillis);
}
if (offset % 2 == 0) {
timeInMillis = timeInMillis + offset;
} else {
timeInMillis = timeInMillis - offset;
}
if (timeInMillis <= 0) {
timeInMillis = offsetInMillis;
}
try {
TimeUnit.MILLISECONDS.sleep(timeInMillis);
} catch (InterruptedException e) {
LOGGER.error("running delay action interrupted", e);
}
}
 

 

    @Override
    public void run(EnhancerModel enhancerModel) throws Exception {
        Exception exception = null;
        String exceptionMessage = null;
        if (exceptionMessageFlag != null) {
            exceptionMessage = enhancerModel.getActionFlag(exceptionMessageFlag.getName());
        }
        if (StringUtil.isBlank(exceptionMessage)) {
            exceptionMessage = DEFAULT_EXCEPTION_MESSAGE;
        }
        if (enhancerModel.getAction().equals(THROW_CUSTOM_EXCEPTION)) {
            exception = throwCustomException(enhancerModel.getClassLoader(), enhancerModel.getActionFlag(exceptionFlag
                .getName()), exceptionMessage);
        } else if (enhancerModel.getAction().equals(THROW_DECLARED_EXCEPTION)) {
            exception = throwDeclaredException(enhancerModel.getClassLoader(), enhancerModel.getMethod(),
                exceptionMessage);
        }
        if (exception != null) {
            // 模拟抛出异常
            InterruptProcessException.throwThrowsImmediately(exception);
        }
    }
   @Override
   public void run(EnhancerModel enhancerModel) throws Exception {
        // get return value from model action
        String value = enhancerModel.getActionFlag(valueFlagSpec.getName());
        Method method = enhancerModel.getMethod();

        if (method == null) {
            return;
        }

        final Map<String, Object> variates = new HashMap<String, Object>();
        if (enhancerModel.getMethodArguments() != null) {
            for (int i = 0; i < enhancerModel.getMethodArguments().length; i++) {
                variates.put(String.format("p%d", i), enhancerModel.getMethodArguments()[i]);
            }
        }
        if (enhancerModel.getReturnValue() != null) {
            variates.put("r", enhancerModel.getReturnValue());
        }

        //这里其实就是一个映射器吧, 翻译基本类型用的
        final Calculator calculator = new Calculator() {
            @Override
            public Constant getValue(String name) throws CompilerException {
                if (name==null || name.equals("null")) {
                    return Constant.build(NULL, null);
                } else if (!variates.containsKey(name)) {
                    return Constant.build(STRING, name);
                } else if (variates.get(name) instanceof Number) {
                    return Constant.build(NUMERIC, ((Number) variates.get(name)).doubleValue());
                } else if (variates.get(name) instanceof String) {
                    return Constant.build(STRING, variates.get(name).toString());
                } else if (variates.get(name) instanceof Boolean) {
                    return Constant.build(BOOLEAN, Boolean.parseBoolean(variates.get(name).toString()));
                }
                return Constant.build(NULL, null);
            }

            @Override
            public boolean isVariate(String name) {
                return variates.containsKey(name);
            }
        };

        final Syntactic syntactic = new Syntactic(calculator);
        final Constant constant = syntactic.getFormulaValue(value);

        // 生成返回值
        Object returnValue = generateReturnValue(enhancerModel.getClassLoader(), method, constant.getAsString());

        InterruptProcessException.throwReturnImmediately(returnValue);
    }
    @Override
    public void run(EnhancerModel enhancerModel) throws Exception {
        if (executorService != null && (!executorService.isShutdown())) {
            throw new IllegalStateException("The jvm cpu full load experiment is running");
        }
        // 绑定 CPU 核心数, 即指定几个核心满载
        String cpuCount = enhancerModel.getActionFlag(JvmConstant.FLAG_NAME_CPU_COUNT);
        // 获取所有核心
        int maxProcessors = Runtime.getRuntime().availableProcessors();
        int threadCount = maxProcessors;
        if (!StringUtil.isBlank(cpuCount)) {
            Integer count = Integer.valueOf(cpuCount.trim());
            if (count > 0 && count < maxProcessors) {
                threadCount = count;
            }
        }
        synchronized (lock) {
            if (executorService != null && (!executorService.isShutdown())) {
                throw new IllegalStateException("The jvm cpu full load experiment is running...");
            }
            // 需要N核心满载,就生成N个的线程,
            executorService = Executors.newFixedThreadPool(threadCount);
        }
        flag = true;
        for (int i = 0; i < threadCount; i++) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    // 死循环, 循环内什么都不要做
                    while (flag) {
                    }
                }
            });
            LOGGER.info("start jvm cpu full load thread, {}", i);
        }
    }
内存这块有三种异常,堆,非堆, 堆外
--堆
@Override
protected void innerRun(EnhancerModel enhancerModel, JvmOomConfiguration jvmOomConfiguration) { oomObjects.add(new OomObject(jvmOomConfiguration.getBlock())); }

--非堆
@Override
protected void innerRun(EnhancerModel enhancerModel, JvmOomConfiguration jvmOomConfiguration) {
Enhancer enhancer = new Enhancer();
enhancer.setSuperclass(OomObject.class);
enhancer.setUseCache(false);
enhancer.setCallback(new MethodInterceptor() {
@Override
public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy)
throws Throwable {
return proxy.invokeSuper(obj, args);
}
});
enhancer.create();

}
--堆外
@Override
protected void innerRun(EnhancerModel enhancerModel, JvmOomConfiguration jvmOomConfiguration) {
oomObjects.add(ByteBuffer.allocateDirect(jvmOomConfiguration.getBlock() * JvmConstant.ONE_MB));
}
 

 

原理就是动态编译生成大量的 class,对编译后的 class 进行实例化并调用。让 JIT 误以为是一个热方法而进行编译。
@Override
public void run() {
            LOGGER.info("Generating all objects for preheating. Bucket size: " + bucketSize + ".");

            List<Object> objects = new ArrayList<Object>();
// 动态类 DynamicJavaClassGenerator generator
= new DynamicJavaClassGenerator(); for (int i = 0; i < bucketSize; i++) { if (!flag.get()) { LOGGER.info("Experiment stopped. Stop code cache object generating."); } Object object = null; // 生成对象 try { object = generator.generateObject(); } catch (Exception e) { LOGGER.error("Generate CodeCacheObject failed.", e); } if (null != object) { objects.add(object); } } LOGGER.info("Generated all objects for code cache filling."); try { lock.await(); } catch (Exception e) { LOGGER.error("Worker thread has been interrupted.", e); return; } LOGGER.info("Preheating all objects. Compile threshold: " + compileThreshold + "."); while (true) { for (Object object : objects) { try { // 不断加热方法,促使进行 JIT编译 Method method = object.getClass().getMethod("method"); for (int i = 0; i <= compileThreshold; i++) { if (!flag.get()) { LOGGER.info("Experiment stopped. Stop code cache object preheating."); return; } method.invoke(object); } } catch (Exception e) { LOGGER.error("Preheating CodeCacheObject failed.", e); } } } } }

以上是关于Java 混沌实验执行器 chaosblade-exec-jvm的主要内容,如果未能解决你的问题,请参考以下文章

面向云原生的混沌工程工具-ChaosBlade

混沌工程之 ChaosToolkit K8S 使用之删除 POD 实验

混沌工程之 ChaosToolkit K8S 使用之删除 POD 实验

为你的服务打针疫苗 —— 混沌工程

为你的服务打针疫苗 —— 混沌工程

ChaosBlade:从混沌工程实验工具到混沌工程平台