java.lang.RuntimeException: ManagedChannel allocation site
	at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:102)
	at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:60)
	at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:51)
	at io.grpc.internal.ManagedChannelImplBuilder.build(ManagedChannelImplBuilder.java:672)
	at io.grpc.ForwardingChannelBuilder2.build(ForwardingChannelBuilder2.java:260)
	at com.vectara.sandbox.controller.PodExecutorManager.watchPods(PodExecutorManager.java:204)
	at java.base/java.lang.Thread.run(Thread.java:1583)


public class PodExecutorManager {
  private static final FluentLogger LOG = FluentLogger.forEnclosingClass();

  private final String uid;
  private final String podName;
  private final String namespace;
  private final String imageTag;
  private final int podPort;
  private final int podCount;
  private final CoreV1Api api;
  private final ApiClient apiClient;
  private final String cpuLimit;
  private final String memoryLimit;
  private final String cpuOverhead;
  private final String memoryOverhead;
  private final Map<String, PodExecutor> podExecutors = new ConcurrentHashMap<>();
  private final ArrayBlockingQueue<PodExecutor> availablePods;
  private final String podTemplate;

  public PodExecutorManager(Builder builder) throws IOException {
    this.uid = requireNonNull(builder.uid, "uid");
    this.podName = requireNonNull(builder.podName, "podName");
    this.namespace = requireNonNull(builder.namespace, "namespace");
    this.imageTag = requireNonNull(builder.imageTag, "imageTag");
    this.podPort = requireNonNull(builder.podPort, "podPort");
    this.podCount = requireNonNull(builder.podCount, "podCount");
    this.availablePods =  new ArrayBlockingQueue<>(podCount);
    this.apiClient = requireNonNull(builder.apiClient, "apiClient");
    this.cpuLimit = requireNonNull(builder.cpuLimit, "cpuLimit");
    this.memoryLimit = requireNonNull(builder.memoryLimit, "memoryLimit");
    this.cpuOverhead = requireNonNull(builder.cpuOverhead, "cpuOverhead");
    this.memoryOverhead = requireNonNull(builder.memoryOverhead, "memoryOverhead");
    api = new CoreV1Api();
    api.setApiClient(apiClient);

    try (InputStream inputStream = PodExecutorManager.class.getClassLoader().getResourceAsStream("pod-template.yaml")) {
      if (inputStream == null) {
        throw new IOException("Resource pod-template.yaml not found");
      }
      podTemplate = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
    }
  }

  private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

  public synchronized PodExecutor getAvailablePod() {
    try {
      PodExecutor pod = availablePods.poll(100, TimeUnit.MILLISECONDS);
      if (pod != null) {
        schedulePodTermination(pod);
      }
      return pod;

    } catch (Exception e) {
      LOG.atWarning().withCause(e).log("Interrupted getting sandbox.");
      return null;
    }
  }

  private void schedulePodTermination(PodExecutor pod) {
    var ctx = Context.current();
    long deadline = Context.current().getDeadline().timeRemaining(TimeUnit.MILLISECONDS);

    if (deadline > 0) {
      scheduler.schedule(() -> {
        if (!ctx.isCancelled()) {
          terminatePod(pod);
        }
      }, deadline, TimeUnit.MILLISECONDS);

    } else {
      LOG.atWarning().log("Deadline has already passed for pod %s", pod.name());
      terminatePod(pod);
    }
  }

  public void terminatePod(PodExecutor pod) {
    LOG.atInfo().log("termination started -> %s", pod.name());
    if (pod.channel() != null && !pod.channel().isShutdown()) {
      try {
        pod.channel().shutdown().awaitTermination(5, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
        LOG.atWarning().withCause(e).log("termination interrupted -> %s", pod.name());
      }
    }
    CompletableFuture.runAsync(() -> terminatePod(pod.name()));
  }

  public void terminatePod(String podName) {
    try {
      api.deleteNamespacedPod(podName, namespace).execute();
    } catch (Exception e) {
      LOG.atSevere().withCause(e).log("Could not delete pod %s.", podName);
    }
  }

  public void start() {
    // Create network policy
    createNetworkPolicy();

    // Start watching pods
    Thread watchThread = new Thread(this::watchPods);
    watchThread.start();

    // Create pods
    for (int i = 0; i < podCount; i++) {
      createPod();
    }

  }

  private void watchPods() {
    LOG.atInfo().log("Started Watching pods for sandbox executor %s", podName);
    while (true) {
      try {
        var selector = "sandbox-executor-owner=%s".formatted(podName);

        // Create a watch for Pods
        try (Watch<V1Pod> watch = Watch.createWatch(
          apiClient,
          api.listNamespacedPod(namespace)
             .labelSelector(selector)
             .watch(true)
             .buildCall(null),
          new TypeToken<Response<V1Pod>>() {
          }.getType())) {

          for (Response<V1Pod> item : watch) {
            if (item.object == null) {
              continue;
            }

            V1Pod pod = item.object;
            String podName = pod.getMetadata().getName();
            String podIp = pod.getStatus().getPodIP();

            switch (item.type) {
              case "ADDED":
              case "MODIFIED":
                switch (pod.getStatus().getPhase()) {
                  case "Running":
                    if (podIp != null) {
                      try {
                        ManagedChannel channel = ManagedChannelBuilder.forAddress(podIp, podPort)
                                                                      .usePlaintext()
                                                                      .build();
                        var executorInfo = new PodExecutor(podName, channel, new CompletableFuture<>());

                        podExecutors.put(podName, executorInfo);
                        availablePods.offer(executorInfo);
                      } catch (RuntimeException e) {
                        LOG.atWarning().withCause(e).log("failed creating channel");
                        CompletableFuture.runAsync(() -> terminatePod(podName));
                      } catch (Exception e) {
                        LOG.atSevere().withCause(e).log("general exception: failed creating channel");
                        CompletableFuture.runAsync(() -> terminatePod(podName));
                      }
                    }
                    break;
                  case "Succeeded":
                  case "Failed":
                  case "Unknown":
                    LOG.atSevere().log("state-> %s",pod.getStatus().toString());
                    terminatePod(podName);
                    break;
                }
                break;
              case "DELETED":
                LOG.atInfo().log("DELETED -> %s", podName);
                var removedExecutor = podExecutors.remove(podName);
                LOG.atInfo().log("removed from map-> %s", podExecutors.size());
                if (removedExecutor.channel() != null && !removedExecutor.channel().isShutdown()) {
                  LOG.atInfo().log("shutting channel -> %s", removedExecutor.name());
                  removedExecutor.channel().shutdown().awaitTermination(5, TimeUnit.SECONDS);
                  removedExecutor.terminationFuture().complete(new PodExecutor.TerminationEvent());
                }
                CompletableFuture.runAsync(this::createPod);
                break;
            }
          }
        } catch (Exception e) {
          LOG.atSevere().withCause(e).log("Error watching pods, retrying.");
        }
      } catch (Exception e) {
        LOG.atWarning().withCause(e).log("Stopped watching pods, retrying.");
      }
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        LOG.atWarning().withCause(e).log("Sleep interrupted, retrying.");
      }
    }
  }