<commit_before>package cluster

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/flynn/flynn/host/types"
	"github.com/flynn/flynn/host/volume"
	"github.com/flynn/flynn/pinkerton/layer"
	"github.com/flynn/flynn/pkg/httpclient"
	"github.com/flynn/flynn/pkg/stream"
)

// Host is a client for a host daemon.
type Host struct {
	id   string
	tags map[string]string
	c    *httpclient.Client
}

// NewHost creates a new Host that uses client to communicate with it.
// addr is used by Attach.
func NewHost(id string, addr string, h *http.Client, tags map[string]string) *Host {
	if h == nil {
		h = http.DefaultClient
	}
	if !strings.HasPrefix(addr, "http") {
		addr = "http://" + addr
	}
	return &Host{
		id:   id,
		tags: tags,
		c: &httpclient.Client{
			ErrNotFound: ErrNotFound,
			URL:         addr,
			HTTP:        h,
		},
	}
}

// ID returns the ID of the host this client communicates with.
func (c *Host) ID() string {
	return c.id
}

// Tags returns the hosts tags
func (c *Host) Tags() map[string]string {
	return c.tags
}

// Addr returns the IP/port that the host API is listening on.
func (c *Host) Addr() string {
	u, err := url.Parse(c.c.URL)
	if err != nil {
		return ""
	}
	return u.Host
}

func (c *Host) GetStatus() (*host.HostStatus, error) {
	var res host.HostStatus
	err := c.c.Get("/host/status", &res)
	return &res, err
}

func WaitForHostStatus(hostIP string, desired func(*host.HostStatus) bool) (*host.HostStatus, error) {
	const waitMax = time.Minute
	const waitInterval = 500 * time.Millisecond
	h := NewHost("", fmt.Sprintf("http://%s:1113", hostIP), nil, nil)
	timeout := time.After(waitMax)
	for {
		status, err := h.GetStatus()
		if err == nil && desired(status) {
			return status, nil
		}
		select {
		case <-timeout:
			if err == nil {
				return nil, fmt.Errorf("desired host status not reached after %s", waitMax)
			}
			return nil, fmt.Errorf("timed out getting host status: %s", err)
		default:
			time.Sleep(waitInterval)
		}
	}
}

// ListJobs lists the jobs running on the host.
func (c *Host) ListJobs() (map[string]host.ActiveJob, error) {
	var jobs map[string]host.ActiveJob
	err := c.c.Get("/host/jobs", &jobs)
	return jobs, err<commit_msg>discoverd,flannel: Wait indefinitely for appropriate host status<commit_after>package cluster

import (
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/flynn/flynn/host/types"
	"github.com/flynn/flynn/host/volume"
	"github.com/flynn/flynn/pinkerton/layer"
	"github.com/flynn/flynn/pkg/httpclient"
	"github.com/flynn/flynn/pkg/stream"
)

// Host is a client for a host daemon.
type Host struct {
	id   string
	tags map[string]string
	c    *httpclient.Client
}

// NewHost creates a new Host that uses client to communicate with it.
// addr is used by Attach.
func NewHost(id string, addr string, h *http.Client, tags map[string]string) *Host {
	if h == nil {
		h = http.DefaultClient
	}
	if !strings.HasPrefix(addr, "http") {
		addr = "http://" + addr
	}
	return &Host{
		id:   id,
		tags: tags,
		c: &httpclient.Client{
			ErrNotFound: ErrNotFound,
			URL:         addr,
			HTTP:        h,
		},
	}
}

// ID returns the ID of the host this client communicates with.
func (c *Host) ID() string {
	return c.id
}

// Tags returns the hosts tags
func (c *Host) Tags() map[string]string {
	return c.tags
}

// Addr returns the IP/port that the host API is listening on.
func (c *Host) Addr() string {
	u, err := url.Parse(c.c.URL)
	if err != nil {
		return ""
	}
	return u.Host
}

func (c *Host) GetStatus() (*host.HostStatus, error) {
	var res host.HostStatus
	err := c.c.Get("/host/status", &res)
	return &res, err
}

func WaitForHostStatus(hostIP string, desired func(*host.HostStatus) bool) (*host.HostStatus, error) {
	const waitInterval = 500 * time.Millisecond
	const logInterval = time.Minute
	start := time.Now()
	lastLogged := time.Now()
	h := NewHost("", fmt.Sprintf("http://%s:1113", hostIP), nil, nil)
	for {
		status, err := h.GetStatus()
		if err == nil && desired(status) {
			return status, nil
		}
		if time.Since(lastLogged) > logInterval {
			log.Printf("desired host status still not reached after %s", time.Since(start))
			lastLogged = time.Now()
		}
		time.Sleep(waitInterval)
	}
}

// ListJobs lists the jobs running on the host.
func (c *Host) ListJobs() (map[string]host.ActiveJob, error) {
	var jobs map[string]host.ActiveJob
	err := c.c.Get("/host/jobs", &jobs)
	return jobs, err
{'fraction_non_alphanumeric': 0.1447721179624665, 'fraction_numerical': 0.003127792672028597, 'mean_word_length': 3.441468253968254, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7015729', 'n_tokens_mistral': 1725, 'n_tokens_neox': 1642, 'n_words': 472}