Pulsar升级自动化：一键搞定集群升级与测试

pulsar-upgrade-cli -h                                                                                                  ok | at 10:33:18 
A cli app for upgrading Pulsar

Usage:
  pulsar-upgrade-cli [command]

Available Commands:
  completion  Generate the autocompletion script for the specified shell
  help        Help about any command
  install     install a target version
  scale       scale statefulSet of the cluster

Flags:
      --burst-limit int                 client-side default throttling limit (default 100)
      --debug                           enable verbose output
  -h, --help                            help for pulsar-upgrade-cli
      --kube-apiserver string           the address and the port for the Kubernetes API server
      --kube-as-group stringArray       group to impersonate for the operation, this flag can be repeated to specify multiple groups.
      --kube-as-user string             username to impersonate for the operation

pulsar-upgrade-cli install \                                                   
        --values ./charts/pulsar/values.yaml \
        --set namespace=pulsar-test \
        --set initialize=true \
        --debug \
        --test-case-schema=http \
        --test-case-host=127.0.0.1 \
        --test-case-port=9999 \
    pulsar-test ./charts/pulsar -n pulsar-test

# 缩容集群规模为0
./pulsar-upgrade-cli scale --replicase 0 -n pulsar-test
# 缩容为最小集群
./pulsar-upgrade-cli scale --replicase 1 -n pulsar-test
# 恢复为最满集群
./pulsar-upgrade-cli scale --replicase 2 -n pulsar-test

require (  
    github.com/spf13/cobra v1.6.1  
    github.com/spf13/pflag v1.0.5   
    helm.sh/helm/v3 v3.10.2
)

func (e *installEvent) FinishInstall(cfg *action.Configuration, name string) error {  
    bar.Increment()  
    bar.Finish()  
  
    clientSet, err := cfg.KubernetesClientSet()  
    if err != nil {  
       return err  
    }  
    ctx := context.Background()  
    ip, err := GetServiceExternalIp(ctx, clientSet, settings.Namespace(), fmt.Sprintf("%s-proxy", name))  
    if err != nil {  
       return err  
    }  
  
    token, err := GetPulsarProxyToken(ctx, clientSet, settings.Namespace(), fmt.Sprintf("%s-token-proxy-admin", name))  
    if err != nil {  
       return err  
    }  
    // trigger testcase  
    err = e.client.Trigger(context.Background(), ip, token)  
    return err  
}

AbstractJobDefine job5 =  
        new FailoverConsumerTest(event, "故障转移消费测试", pulsarClient, 20, admin);  
CompletableFuture<Void> c5 = CompletableFuture.runAsync(job5::start, EXECUTOR);  
AbstractJobDefine job6 = new SchemaTest(event,"schema测试",pulsarClient,20,prestoService);  
CompletableFuture<Void> c6 = CompletableFuture.runAsync(job6::start, EXECUTOR);  
AbstractJobDefine job7 = new VlogsTest(event,"vlogs test",pulsarClient,20, vlogsUrl);  
CompletableFuture<Void> c7 = CompletableFuture.runAsync(job7::start, EXECUTOR);  
  
CompletableFuture<Void> all = CompletableFuture.allOf(c1, c2, c3, c4, c5, c6, c7);  
all.whenComplete((___, __) -> {  
    event.finishAll();  
    pulsarClient.closeAsync();  
    admin.close();  
}).get();

@Data
public abstract class AbstractJobDefine {
    private Event event;
    private String jobName;
    private PulsarClient pulsarClient;

    private int timeout;

    private PulsarAdmin admin;

    public AbstractJobDefine(Event event, String jobName, PulsarClient pulsarClient, int timeout, PulsarAdmin admin) {
        this.event = event;
        this.jobName = jobName;
        this.pulsarClient = pulsarClient;
        this.timeout = timeout;
        this.admin = admin;
    }

    public void start() {
        event.addJob();
        try {
            CompletableFuture.runAsync(() -> {
                StopWatch watch = new StopWatch();
                try {
                    watch.start(jobName);
                    run(pulsarClient, admin);
                } catch (Exception e) {
                    event.oneException(this, e);
                } finally {
                    watch.stop();
                    event.finishOne(jobName, StrUtil.format("cost: {}s", watch.getTotalTimeSeconds()));
                }
            }, TestCase.EXECUTOR).get(timeout, TimeUnit.SECONDS);
        } catch (Exception e) {
            event.oneException(this, e);
        }
    }


    /** run busy code
     * @param pulsarClient pulsar client
     * @param admin pulsar admin client
     * @throws Exception e
     */
    public abstract void run(PulsarClient pulsarClient, PulsarAdmin admin) throws Exception;
}

public interface Event {  
  
    /**  
     * 新增一个任务  
     */  
    void addJob();  
  
    /** 获取运行中的任务数量  
     * @return 获取运行中的任务数量  
     */  
    TestCaseRuntimeResponse getRuntime();  
  
    /**  
     * 单个任务执行完毕  
     *  
     * @param jobName    任务名称  
     * @param finishCost 任务完成耗时  
     */  
    void finishOne(String jobName, String finishCost);  
  
    /**单个任务执行异常  
     * @param jobDefine 任务  
     * @param e 异常  
     */  
    void oneException(AbstractJobDefine jobDefine, Exception e);  
  
    /**  
     * 所有任务执行完毕  
     */  
    void finishAll();  
}