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
下面依次展开

1. 线程池被占用

 @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);
}

2. 连接池被占用

@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);
}

3. 方法延迟

@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);
    }
}
 

 

4. 抛异常

    @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);
        }
    }

5. 指定返回值

   @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);
    }

6. CPU 满载

    @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);
        }
    }

7. 内存溢出

内存这块有三种异常，堆，非堆， 堆外
--堆
@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));
}
 

 

8. CodeCacheFilling

原理就是动态编译生成大量的 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);
                    }
                }
            }
        }

    }

后记

个人注解版：https://github.com/zhaoyb/chaosblade-exec-jvm

上面的只是混沌工程的一部分，完整部分chaosblade (https://github.com/chaosblade-io/chaosblade)，可以完成对操作系统的异常注入，比如 CPU，内存，硬盘，网络。关于这一块参考：https://blog.csdn.net/u013256816/article/details/99917021

 

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

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

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

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

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

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

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