經過Hack方式實現SDC中Stage配置聯動刷新

時間 2019-11-06

標籤經過 hack 方式實現 sdc stage 配置聯動刷新简体版

原文原文鏈接

問題描述

最近項目組準備開發一個IoT平臺項目，須要使用到StreamSets DataCollector組件進行數據處理。
其中的一個Stage，產品經理設計了一個以下的配置界面：
前端

預期的展現效果是經過下拉「物實例」列表框的時候，根據所選擇物實例的屬性個數聯動刷新「屬性匹配」，並且物實例下拉框的數據是經過API獲取的。java

這帶來2個問題：git

如何實現下拉框列表中的數據從外部獲取？
如何實現根據所選下拉框數據聯動刷新「屬性匹配」的界面？

實際上，單純的下拉列表和聯動刷新SDC是原生支持的，可是下拉列表的數據是靜態配置的，並且聯動刷新的界面也是預先配置的。而咱們的項目需求是須要根據下拉列表中選擇的物實例屬性個數進行聯動刷新，而不一樣的物實例的屬性個數並不相同，所以沒法作到預先配置。後端

因此，咱們的原型設計SDC原生並不能支持。
可是產品設計並不但願修改，所以只能尋找對應的解決辦法。api

如何從外部獲取下拉列表參數

對於下拉列表的數據從外部獲取這個實現相對容易，在Stage中對於下拉列表的配置一般使用以下方式：瀏覽器

// 物實例下拉列表
@ConfigDef(
    required = true,
    type = ConfigDef.Type.MODEL,
    label = "Instance",
    defaultValue = "",
    displayPosition = 30,
    group = "DIGITALTWIN",
    description = "Instance List"
)
@ValueChooserModel(DigitalTwinInstanceChooser.class)
public String instance = null;

其中，DigitalTwinInstanceChooser類是數據源，它必須實現接口com.streamsets.pipeline.api.ChooserValues，即必須實現以下接口方法：架構

public interface ChooserValues {
    String getResourceBundle();

    List<String> getValues();

    List<String> getLabels();
}

其中，方法getResourceBundle()是資源國際化配置參數，getValues()爲下拉列表選項中各項對應的value，getLabels()爲下拉列表選項中各項在界面上顯示的key。所以，爲了實現下拉列表數據從外部獲取，只須要在實現了接口ChooserValues的類構造方法中初始化對應數據便可，以下示例：框架

public class DigitalTwinInstanceChooser implements ChooserValues {
    private static List<String> values = null;
    private static List<String> labels = null;
    public DigitalTwinInstanceChooser() {
        // 只須要刷新一次
        if(values != null) {
            return;
        }
        List<DigitalTwinInstance> list = DigitalTwinStreamClient.getDigitalTwinInstanceList();
        setList(list);
    }

    // 數據初始化
    public static void setList(List<DigitalTwinInstance> list) {
        if(list == null) {
            return;
        }
        values = new ArrayList<String>(list.size());
        labels = new ArrayList<String>(list.size());

        for(DigitalTwinInstance dtb : list) {
            if(dtb == null) {
                continue;
            }
            values.add(dtb.getId()+ "");
            labels.add(dtb.getName());
        }
    }

    @Override
    public String getResourceBundle() {
        return null;
    }

    @Override
    public List<String> getValues() {
        return values;
    }

    @Override
    public List<String> getLabels() {
        return labels;
    }
}

如何實現根據下拉列表選項動態刷新

在咱們的這個項目需求中是須要根據下拉選中的物實例屬性個數動態刷新界面的，這個在SDC中原生並不支持。一開始我沒有任何思路，組裏熟悉這個框架的人問了一圈，沒一我的解決過相似問題。其中一位同事告訴我以前一位已經離職的同事遇到過相似的需求，可是具體怎麼實現的並不清楚，只是說好像經過修改前端來完成的。雖然這個信息沒有直接解決個人問題，可是卻給我打開了一點思路。咱們知道，在SDC的Stage配置中是實時保存的。SDC的前端使用AugularJS框架，只要用戶配置參數發生了變化，就會實時經過API保存到後端，這樣Stage在運行時就能獲取到用戶配置的對應參數。順着這個思路，我對Stage保存參數的請求進行了抓包，通過對每一次保存請求參數和API接口的返回結果進行對比發現：前端每一次將保存參數經過API發送到後臺進行保存以後會將該參數再返回給前端。因而我就腦洞大開：之因此須要將用戶設置的參數再返回給前端，應該是前端須要這些參數進行界面渲染。那麼，對於我這個需求，當用戶選擇了某個具體的物實例以後，是否能夠在後端根據傳遞的物實例參數動態將對應的屬性參數返回給前端，這樣前端就能夠動態渲染出相應的「屬性匹配」界面了呢？可是這樣的話就須要修改SDC保存Stage配置參數的源碼了，報着試一試的心態因而開始了以下Hack實踐。
第一步，找到保存Stage參數的API接口。在瀏覽器中能夠看到，保存Stage配置參數的地址爲：/rest/v1/pipeline/{pipelineid}，因而憑直接找到了對應API接口類：datacollector\container\src\main\java\com\streamsets\datacollector\restapi\PipelineStoreResource.java，在該接口中有一個更新Pipeline的方法：ide

@Path("/pipeline/{pipelineId}")
@POST
@ApiOperation(value = "Update an existing Pipeline Configuration by name", response = PipelineConfigurationJson.class,
    authorizations = @Authorization(value = "basic"))
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    @RolesAllowed({
        AuthzRole.CREATOR, AuthzRole.ADMIN, AuthzRole.CREATOR_REMOTE, AuthzRole.ADMIN_REMOTE
    })
    public Response savePipeline(
        @PathParam("pipelineId") String name,
        @QueryParam("rev") @DefaultValue("0") String rev,
        @QueryParam("description") String description,
        @ApiParam(name="pipeline", required = true) PipelineConfigurationJson pipeline) throws  URISyntaxException, PipelineException {
        if (store.isRemotePipeline(name, rev)) {
            throw new PipelineException(ContainerError.CONTAINER_01101, "SAVE_PIPELINE", name);
        }
        PipelineInfo pipelineInfo = store.getInfo(name);
        RestAPIUtils.injectPipelineInMDC(pipelineInfo.getTitle(), pipelineInfo.getPipelineId());
        PipelineConfiguration pipelineConfig = BeanHelper.unwrapPipelineConfiguration(pipeline);
        PipelineConfigurationValidator validator = new PipelineConfigurationValidator(stageLibrary, name, pipelineConfig);
        pipelineConfig = validator.validate();

        // 在這裏判斷並處理用戶當前返回的配置
        // 1.檢查下拉菜單的值是否發生變化
    if(checkDtInstanceChanged(pipelineInfo, rev, pipelineConfig)) {
        LOG.info("Need update DT Instance attribute!");
        // 2.若是下拉菜單的值發生了變化才動態返回值
        pipelineConfig = updateDigitalTwinConfig(pipelineConfig);
    }

    pipelineConfig = store.save(user, name, rev, description, pipelineConfig);
    return Response.ok().entity(BeanHelper.wrapPipelineConfiguration(pipelineConfig)).build();
}

第二步，在接口方法中根據需求實現對應的邏輯，動態返回下拉列表中選擇物實例信息。ui

private boolean checkDtInstanceChanged(PipelineInfo pipelineInfo, String rev, PipelineConfiguration pipelineConfig) {
    try {
        // 加載存儲的數據
        PipelineConfiguration storePipeLine = store.load(pipelineInfo.getPipelineId(), rev);
        // 讀取已經存儲的配置參數
        String storeDtInstance = getDtInstanceId(storePipeLine);
        // 讀取前端發送過來的配置參數
        String paramDtInstance = getDtInstanceId(pipelineConfig);
        LOG.info("Check DT Instance Changed, stored: {}, param: {}", storeDtInstance, paramDtInstance);
        return !storeDtInstance.equals(paramDtInstance);
    } catch (PipelineException e) {
      e.printStackTrace();
    }
    return false;
  }

  // 從配置中讀取參數
  private String getDtInstanceId(PipelineConfiguration pipeline) {
      if(pipeline == null) {
          return "";
      }

      List<StageConfiguration> stageList = pipeline.getStages();
      for(StageConfiguration stage : stageList) {
          if(!"digitaltwin-stage".equals(stage.getLibrary())) {
              continue;
          }
          List<Config> configList = stage.getConfiguration();
          for(Config config : configList) {
              if(!"config.instance".equals(config.getName())) {
                  continue;
              }
              Object value = config.getValue();
              if(value == null) {
                return "";
              }
              return value.toString();
          }
      }
      return "";
  }

  // 實現動態更新前端界面
  private PipelineConfiguration updateDigitalTwinConfig(PipelineConfiguration pipelineConfig) {
      List<StageConfiguration> stages = pipelineConfig.getStages();
      if(stages == null || stages.isEmpty()) {
          return pipelineConfig;
      }

      for(StageConfiguration stage : stages) {
          if(stage == null) {
              continue;
          }
          if(!"digitaltwin-stage".equals(stage.getLibrary())) {
              continue;
          }

          String resourceURL = null;
          String instance = null;
          List<Config> configList = stage.getConfiguration();
          int size = configList.size();
          List<Config> newConfigList = new ArrayList<Config>(size);
          int index = -1;
          String key = "config.attrs";
          for(int i = 0; i < size; i++) {
              Config config = configList.get(i);
              if(config == null) {
                  continue;
              }
              if(key.equals(config.getName())) {
                  index = i;
                  newConfigList.add(null);
                  continue;
              }
              if("config.resourceURL".equals(config.getName())) {
                  resourceURL = config.getValue().toString();
              }
              if("config.instance".equals(config.getName())) {
                  instance = config.getValue().toString();
              }
              newConfigList.add(config);
          }
          if((resourceURL == null || "".equals(resourceURL.trim())
                  || (instance == null || "".equals(instance.trim())))) {
              return pipelineConfig;
          }

          // 動態讀取屬性列表
          List<DtConfig> list = DtStreamClient.getDtInstanceAttrList(resourceURL, instance);
          //newConfigList.add(new Config("config.attrs", list));
          newConfigList.set(index, new Config(key, list));
          stage.setConfig(newConfigList);

          // 強制前端刷新界面
          /*Map<String, Object> uiInfo = stage.getUiInfo();
          try {
              String key = "xPos";
              Object xPos = uiInfo.get(key);
              if(xPos != null) {
                  int x = Integer.valueOf(xPos.toString());
                  x += 1;
                  uiInfo.put(key, x);
              }
          } catch (NumberFormatException e) {
              e.printStackTrace();
          }*/
      }

      return pipelineConfig;
}

最後進行驗證，居然成功了！！！

總結

Stage中關於「屬性匹配」參數類型必須爲Map結構類型：type = ConfigDef.Type.MAP，以下：

// 動態切換屬性
@ConfigDef(
        required = false,
        type = ConfigDef.Type.MAP,
        label = "DigitalTwin Attribute Map",
        displayPosition = 40,
        group = "DIGITALTWIN"
)
public Map<String, String> attrs = new HashMap<String, String>();
public Map<String, String> getAttrs() {
    return this.attrs;
}