Spinnaker第七节—Orca代码详解

程序员文章站 2022-04-16 23:36:16

...

这个春节假期由于新冠弄得一直在家憋着，闲得无事把spinnaker代码翻出来把每个微服务重新review一遍，做了一些笔记，今天拿来分享一下，后续如果有spinnaker开发或者bug修复可以直接有的放矢。计划先把核心的几个微服务分几期慢慢介绍下：orca、clouddriver、deck，今天先介绍orca。

Orca名词解释

Orchestration
   与pipeline平行，我个人理解是orca这个微服务主要是为了支持pipeline工作流的，为了对接临时性的任务包扩展出了ORCHESTRATION
   Execution
   一次执行。分为两种类型:PIPELINE和ORCHESTRATION，核心属性数一个stage的集合
   Stage
   orca最复杂的类型，关联一个execution和多个task，还维护有parentStage、downStreamStage、output等重要信息
   StageContext
   驱动这个阶段的环境，提供组件步骤所需的输入
   Task
   Stage的组成部分，orca中最小的操作单元
   ExecutionRunner Execution的执行者
   ExecutionRepository Execution的存储者，我们的是redis，也承载着检索查询
ExecutionLauncher Execution的启动者，接收json封装成Execution，进行存储和启动
   1 Json封装Execution也会分为PIPELINE和ORCHESTRATION两种，本质没区别
   2 最终启动执行：

	    public Execution start(Execution execution) throws Exception {
			executionRunner.start(execution);
			return execution;
		}

orca-web子项目（入口）

controller
OperationsController：与Execution相关的controller
   /orchestrate*
   最终startPipeline，以PIPELINE形式启动execution
   /plan*
   不会真正执行，经过一系列校验后通过ExecutionRepository存储executor
   /ops*
   最终startTask，以ORCHESTRATION形式启动executor
TaskController：
   /pipelines*
   对于pipeline的停、重、查，最终映射的对象类型还是execution
   /task*
   对于task的停、重、查，最终映射的对象类型还是execution
   orca-web总结：OperationsController和TaskController看似有重叠的部分都会涉及到execution，但是侧重点不同。OperationsController侧重于创建和启动，TaskController侧重于后续的暂停、撤销和查询。

orca-core子项目（抽象层）

根目录：
定义一系列的task：retryableTask、cancelableTask等
annotations 自定义sync，用于标注同步，可以作为自定义标注的学习参考
clouddriver.utils 根据stage获取各种clouddriver的信息
commands 意义不大，不需要看
config：orca配置
exceptions：除开定义了异常，还定义了处理异常的Handler，借鉴下

/*
 * Copyright 2014 Netflix, Inc.
 *
 * 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 com.netflix.spinnaker.orca.exceptions;

import java.util.Collections;
import java.util.Map;
import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
import static java.lang.String.format;
import static org.springframework.core.Ordered.LOWEST_PRECEDENCE;

@Order(LOWEST_PRECEDENCE)
public class DefaultExceptionHandler implements ExceptionHandler {
  private final Logger log = LoggerFactory.getLogger(getClass());

  @Override public boolean handles(Exception e) {
    return true;
  }

  @Override
  public ExceptionHandler.Response handle(String taskName, Exception e) {
    Map<String, Object> exceptionDetails = ExceptionHandler.responseDetails("Unexpected Task Failure", Collections.singletonList(e.getMessage()));
    exceptionDetails.put("stackTrace", Throwables.getStackTraceAsString(e));
    log.warn(format("Error occurred during task %s", taskName), e);
    return new ExceptionHandler.Response(e.getClass().getSimpleName(), taskName, exceptionDetails, false);
}
}

events：
定义了execution、stage、task的各种事件，
也定了以上事件的监听基类，本质上没有自己做处理，都是丢给delegate在处理，代理模式

  private void onTaskStarted(TaskStarted event) {
    Execution execution = retrieve(event);
    List<Stage> stages = execution.getStages();
    stages
      .stream()
      .filter(it -> it.getId().equals(event.getStageId()))
      .findFirst()
      .ifPresent(stage ->
        delegate.beforeTask(
          persister,
          stage,
          stage.getTasks().stream().filter(it -> it.getId().equals(event.getTaskId())).findFirst().get()
        )
      );
  }

listeners：
定义了ExecutionListener和StageListener两个接口，也就是events中的delegate，定义了一些阶段和要重写的方法。像echo通知等就是这种实现机制。

/*
 * Copyright 2016 Netflix, Inc.
 *
 * 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 com.netflix.spinnaker.orca.listeners;

import com.netflix.spinnaker.orca.ExecutionStatus;
import com.netflix.spinnaker.orca.pipeline.model.Stage;
import com.netflix.spinnaker.orca.pipeline.model.Task;

public interface StageListener {
  default void beforeTask(Persister persister,
                                                Stage stage,
                                                Task task) {
    // do nothing
  }

  default void beforeStage(Persister persister,
                                                 Stage stage) {
    // do nothing
  }

  default void afterTask(Persister persister,
                                               Stage stage,
                                               Task task,
                                               ExecutionStatus executionStatus,
                                               boolean wasSuccessful) {
    // do nothing
  }

  default void afterStage(Persister persister,
                                                Stage stage) {
    // do nothing
  }
}

locks
看了下是为了多节点部署而设计的，目前阶段可以不关心
pipeline
最核心的一个类StageDefinitionBuilder，所有的stage都要继承它
AcquireLockStage、CheckPreconditionsStage、EvaluateVariablesStage、ReleaseLockStage、WaitStage

orca-clouddriver子项目（核心实现）

先搞明白3个服务：oort、mort和kato，这三个都是指向了clouddriver的restful
   oort是大头：对接cluster、image、instance、loadbalance等云平台资源
   mort是小头：对接vpc、安全组、账号鉴权
   kkato是辅助：对接其它辅助的资源
   pipeline：
   stage对于task的组装，注意其中的provider包是留给厂商自己私有化的，所以关于orca云厂商差异化的部分也都集中在这里。
   tasks：
   具体的task实现，也是未来集中关心的地方，provider中各个厂商都有自己的私有化

Spinnaker第七节—Orca代码详解