code: def non_terminated_nodes(
      self, tag_filters: Dict[str, str]
  ) -> List[str]:
      tags = [
          {
              "Key": TAG_RAY_CLUSTER_NAME,
              "Value": self.cluster_name,
          },
      ]
      for k, v in tag_filters.items():
          tags.append(
              {
                  "Key": k,
                  "Value": v,
              }
          )
  
      instances = self.acs.describe_instances(tags=tags)
      non_terminated_instance = []
      for instance in instances:
          if (
              instance.get("Status") == RUNNING
              or instance.get("Status") == PENDING
          ):
              non_terminated_instance.append(instance.get("InstanceId"))
              self.cached_nodes[instance.get("InstanceId")] = instance
      return non_terminated_instance
  
imports: import logging
  import random
  import threading
  import time
  from collections import defaultdict
  from typing import Any, Dict, List, Optional
  from ray.autoscaler._private.aliyun.config import PENDING, RUNNING, STOPPED, STOPPING, bootstrap_aliyun
  from ray.autoscaler._private.aliyun.utils import AcsClient
  from ray.autoscaler._private.cli_logger import cli_logger
  from ray.autoscaler._private.constants import BOTO_MAX_RETRIES
  from ray.autoscaler._private.log_timer import LogTimer
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, TAG_RAY_LAUNCH_CONFIG, TAG_RAY_NODE_KIND, TAG_RAY_NODE_NAME, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 2 for calling the AliyunNodeProvider obj's underlying member method code with example usage: obj.non_terminated_nodes(tag_filters) and returns: non_terminated_instance