@Override
public Map<String, Object> create(Map<String, Object> job) {

    if(Introspection.resolve(job, Fields.METADATA_NAME) != null) {
        throw new RuntimeException("The job instance must contains a 'metadata.name' path");
    }

    V1Job k8sJob = new V1Job();
    k8sJob.setApiVersion(Defaults.API_VERSION);
    k8sJob.setKind(Defaults.KIND);

    V1ObjectMeta k8sMetaData = new V1ObjectMeta();
    k8sMetaData.setName(Introspection.resolve(job, Fields.METADATA_NAME));
    k8sJob.setMetadata(k8sMetaData);

    V1JobSpec k8sJobSpec = new V1JobSpec();


    return null;
}
 

@Before
public void initialize() throws IOException {
  final V1ServiceStatus v1ServiceStatus =
      new V1ServiceStatus()
          .loadBalancer(
              new V1LoadBalancerStatusBuilder()
                  .addToIngress(
                      new V1LoadBalancerIngressBuilder().withIp(detectedAdvertisedHost).build())
                  .build());
  when(v1Service.getStatus()).thenReturn(v1ServiceStatus);
  when(v1Service.getSpec())
      .thenReturn(
          new V1ServiceSpec()
              .ports(
                  Arrays.asList(
                      new V1ServicePort()
                          .name(NatServiceType.JSON_RPC.getValue())
                          .port(rpcHttpPort)
                          .targetPort(new IntOrString(rpcHttpPort)),
                      new V1ServicePort()
                          .name(NatServiceType.RLPX.getValue())
                          .port(p2pPort)
                          .targetPort(new IntOrString(p2pPort)),
                      new V1ServicePort()
                          .name(NatServiceType.DISCOVERY.getValue())
                          .port(p2pPort)
                          .targetPort(new IntOrString(p2pPort)))));
  when(v1Service.getMetadata()).thenReturn(new V1ObjectMeta().name(DEFAULT_BESU_POD_NAME_FILTER));
  natManager = new KubernetesNatManager(DEFAULT_BESU_POD_NAME_FILTER);
  try {
    natManager.start();
  } catch (Exception ignored) {
    System.err.println("Ignored missing Kube config file in testing context.");
  }
  natManager.updateUsingBesuService(v1Service);
}
 

@VisibleForTesting
V1Service createService() {
    final String jobName = createJobName(instanceConfig.getFunctionDetails());

    final V1Service service = new V1Service();

    // setup stateful set metadata
    final V1ObjectMeta objectMeta = new V1ObjectMeta();
    objectMeta.name(jobName);
    objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails()));
    // we don't technically need to set this, but it is useful for testing
    objectMeta.setNamespace(jobNamespace);
    service.metadata(objectMeta);

    // create the stateful set spec
    final V1ServiceSpec serviceSpec = new V1ServiceSpec();

    serviceSpec.clusterIP("None");

    final V1ServicePort servicePort = new V1ServicePort();
    servicePort.name("grpc").port(grpcPort).protocol("TCP");
    serviceSpec.addPortsItem(servicePort);

    serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails()));

    service.spec(serviceSpec);

    // let the customizer run but ensure it doesn't change the name so we can find it again
    final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service);
    overridden.getMetadata().name(jobName);

    return overridden;
}
 

private static V1ObjectMeta parseMetadata(ExperimentSpec experimentSpec) {
  V1ObjectMeta meta = new V1ObjectMeta();
  meta.setName(experimentSpec.getMeta().getName());
  meta.setNamespace(experimentSpec.getMeta().getNamespace());
  return meta;
}
 

private void validateMetadata(ExperimentMeta expectedMeta, V1ObjectMeta actualMeta,
    String actualFramework) {
  Assert.assertEquals(expectedMeta.getName(), actualMeta.getName());
  Assert.assertEquals(expectedMeta.getNamespace(), actualMeta.getNamespace());
  Assert.assertEquals(expectedMeta.getFramework().toLowerCase(), actualFramework);
}
 

@VisibleForTesting
V1StatefulSet createStatefulSet() {
    final String jobName = createJobName(instanceConfig.getFunctionDetails());

    final V1StatefulSet statefulSet = new V1StatefulSet();

    // setup stateful set metadata
    final V1ObjectMeta objectMeta = new V1ObjectMeta();
    objectMeta.name(jobName);
    objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails()));
    // we don't technically need to set this, but it is useful for testing
    objectMeta.setNamespace(jobNamespace);
    statefulSet.metadata(objectMeta);

    // create the stateful set spec
    final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec();
    statefulSetSpec.serviceName(jobName);
    statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism());

    // Parallel pod management tells the StatefulSet controller to launch or terminate
    // all Pods in parallel, and not to wait for Pods to become Running and Ready or completely
    // terminated prior to launching or terminating another Pod.
    statefulSetSpec.setPodManagementPolicy("Parallel");

    // add selector match labels
    // so the we know which pods to manage
    final V1LabelSelector selector = new V1LabelSelector();
    selector.matchLabels(getLabels(instanceConfig.getFunctionDetails()));
    statefulSetSpec.selector(selector);

    // create a pod template
    final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec();

    // set up pod meta
    final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails()));
    templateMetaData.annotations(getPrometheusAnnotations());
    podTemplateSpec.setMetadata(templateMetaData);

    final List<String> command = getExecutorCommand();
    podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null));

    statefulSetSpec.setTemplate(podTemplateSpec);

    statefulSet.spec(statefulSetSpec);

    // let the customizer run but ensure it doesn't change the name so we can find it again
    final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet);
    overridden.getMetadata().name(jobName);

    return statefulSet;
}
 

/**
 * Get the metadata
 *
 * @return meta
 */
public V1ObjectMeta getMetadata() {
  return metadata;
}
 

/**
 * Set metadata
 *
 * @param metadata meta
 */
public void setMetadata(V1ObjectMeta metadata) {
  this.metadata = metadata;
}