def _get_sniff_data()

in opensearchpy/transport.py [0:0]

• 26 lines of code
• 9 McCabe index (conditional complexity)

    def _get_sniff_data(self, initial=False):
        """
        Perform the request to get sniffing information. Returns a list of
        dictionaries (one per node) containing all the information from the
        cluster.

        It also sets the last_sniff attribute in case of a successful attempt.

        In rare cases it might be possible to override this method in your
        custom Transport class to serve data from alternative source like
        configuration management.
        """
        previous_sniff = self.last_sniff

        try:
            # reset last_sniff timestamp
            self.last_sniff = time.time()
            # go through all current connections as well as the
            # seed_connections for good measure
            for c in chain(self.connection_pool.connections, self.seed_connections):
                try:
                    # use small timeout for the sniffing request, should be a fast api call
                    _, headers, node_info = c.perform_request(
                        "GET",
                        "/_nodes/_all/http",
                        timeout=self.sniff_timeout if not initial else None,
                    )

                    # Lowercase all the header names for consistency in accessing them.
                    headers = {
                        header.lower(): value for header, value in headers.items()
                    }

                    node_info = self.deserializer.loads(
                        node_info, headers.get("content-type")
                    )
                    break
                except (ConnectionError, SerializationError):
                    pass
            else:
                raise TransportError("N/A", "Unable to sniff hosts.")
        except Exception:
            # keep the previous value on error
            self.last_sniff = previous_sniff
            raise

        return list(node_info["nodes"].values())