class_docstr: A node provider that implements provider for nodes of Ray on spark.
class_docstr_tok: ["A","node","provider","that","implements","provider","for","nodes","of","Ray","on","spark","."]
code: def non_terminated_nodes(self, tag_filters):
      with self.lock:
          nodes = []
  
          died_nodes = []
          for node_id in self._nodes:
              if node_id == str(HEAD_NODE_ID):
                  status = "running"
              else:
                  status = self._query_node_status(node_id)
  
              if status == "running":
                  if (
                      self._nodes[node_id]["tags"][TAG_RAY_NODE_STATUS]
                      == STATUS_SETTING_UP
                  ):
                      self._nodes[node_id]["tags"][
                          TAG_RAY_NODE_STATUS
                      ] = STATUS_UP_TO_DATE
                      logger.info(
                          f"Spark node provider node {node_id} starts running."
                      )
  
              if status == "terminated":
                  died_nodes.append(node_id)
              else:
                  tags = self.node_tags(node_id)
                  ok = True
                  for k, v in tag_filters.items():
                      if tags.get(k) != v:
                          ok = False
                  if ok:
                      nodes.append(node_id)
  
          for died_node_id in died_nodes:
              self._nodes.pop(died_node_id)
  
          return nodes
  
imports: import json
  import logging
  import sys
  from threading import RLock
  from typing import Any, Dict, Optional
  import requests
  from ray.autoscaler.node_launch_exception import NodeLaunchException
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import NODE_KIND_HEAD, NODE_KIND_WORKER, STATUS_SETTING_UP, STATUS_UP_TO_DATE, TAG_RAY_NODE_KIND, TAG_RAY_NODE_NAME, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 3 for calling the SparkNodeProvider obj's underlying member method code with example usage: obj.non_terminated_nodes(tag_filters) and returns: nodes