blob: e49403aead3ceb993c2b40d790141c52ae42f455 [file] [log] [blame]
// Copyright 2015 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package kubernetes contains a minimal client for the Kubernetes API.
package kubernetes
import (
"bufio"
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"net/url"
"os"
"strings"
"time"
"golang.org/x/build/kubernetes/api"
"golang.org/x/net/context/ctxhttp"
)
// Client is a client for the Kubernetes master.
type Client struct {
httpClient *http.Client
// endPointURL is the Kubernetes master URL ending in
// "/api/v1".
endpointURL string
namespace string // always in URL path-escaped form (for now)
}
// NewClient returns a new Kubernetes client.
// The provided host is an url (scheme://hostname[:port]) of a
// Kubernetes master without any path.
// The provided client is an authorized http.Client used to perform requests to the Kubernetes API master.
func NewClient(baseURL, namespace string, client *http.Client) (*Client, error) {
if namespace == "" {
return nil, fmt.Errorf("must specify Kubernetes namespace")
}
validURL, err := url.Parse(baseURL)
if err != nil {
return nil, fmt.Errorf("failed to parse URL %q: %v", baseURL, err)
}
return &Client{
endpointURL: strings.TrimSuffix(validURL.String(), "/") + "/api/v1",
httpClient: client,
namespace: namespace,
}, nil
}
// Close closes any idle HTTP connections still connected to the Kubernetes master.
func (c *Client) Close() error {
if tr, ok := c.httpClient.Transport.(*http.Transport); ok {
tr.CloseIdleConnections()
}
return nil
}
// nsEndpoint returns the API endpoint root for this client.
// (This has nothing to do with Service Endpoints.)
func (c *Client) nsEndpoint() string {
return c.endpointURL + "/namespaces/" + c.namespace + "/"
}
// RunLongLivedPod creates a new pod resource in the default pod namespace with
// the given pod API specification. It assumes the pod runs a
// long-lived server (i.e. if the container exit quickly, even
// with success, then that is an error).
//
// It returns the pod status once it has entered the Running phase.
// An error is returned if the pod can not be created, or if ctx.Done
// is closed.
func (c *Client) RunLongLivedPod(ctx context.Context, pod *api.Pod) (*api.PodStatus, error) {
var podJSON bytes.Buffer
if err := json.NewEncoder(&podJSON).Encode(pod); err != nil {
return nil, fmt.Errorf("failed to encode pod in json: %v", err)
}
postURL := c.nsEndpoint() + "pods"
req, err := http.NewRequest("POST", postURL, &podJSON)
if err != nil {
return nil, fmt.Errorf("failed to create request: POST %q : %v", postURL, err)
}
res, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return nil, fmt.Errorf("failed to make request: POST %q: %v", postURL, err)
}
body, err := io.ReadAll(res.Body)
res.Body.Close()
if err != nil {
return nil, fmt.Errorf("failed to read request body for POST %q: %v", postURL, err)
}
if res.StatusCode != http.StatusCreated {
return nil, fmt.Errorf("http error: %d POST %q: %q: %v", res.StatusCode, postURL, string(body), err)
}
var podResult api.Pod
if err := json.Unmarshal(body, &podResult); err != nil {
return nil, fmt.Errorf("failed to decode pod resources: %v", err)
}
for {
// TODO(bradfitz,evanbrown): pass podResult.ObjectMeta.ResourceVersion to PodStatus?
ps, err := c.PodStatus(ctx, podResult.Name)
if err != nil {
return nil, err
}
switch ps.Phase {
case api.PodPending:
// The main phase we're waiting on
break
case api.PodRunning:
return ps, nil
case api.PodSucceeded, api.PodFailed:
return nil, fmt.Errorf("pod entered phase %q", ps.Phase)
default:
log.Printf("RunLongLivedPod poll loop: pod %q in unexpected phase %q; sleeping", podResult.Name, ps.Phase)
}
select {
case <-time.After(5 * time.Second):
case <-ctx.Done():
// The pod did not leave the pending
// state. Try to clean it up.
go c.DeletePod(context.Background(), podResult.Name)
return nil, ctx.Err()
}
}
}
func (c *Client) do(ctx context.Context, method, urlStr string, dst interface{}) error {
req, err := http.NewRequest(method, urlStr, nil)
if err != nil {
return err
}
res, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return err
}
defer res.Body.Close()
if res.StatusCode != http.StatusOK {
body, _ := io.ReadAll(res.Body)
return fmt.Errorf("%v %s: %v, %s", method, urlStr, res.Status, body)
}
if dst != nil {
var r io.Reader = res.Body
if false && strings.Contains(urlStr, "endpoints") { // for debugging
r = io.TeeReader(r, os.Stderr)
}
return json.NewDecoder(r).Decode(dst)
}
return nil
}
// GetServices returns all services in the cluster, regardless of status.
func (c *Client) GetServices(ctx context.Context) ([]api.Service, error) {
var list api.ServiceList
if err := c.do(ctx, "GET", c.nsEndpoint()+"services", &list); err != nil {
return nil, err
}
return list.Items, nil
}
// Endpoint represents a service endpoint address.
type Endpoint struct {
IP string
Port int
PortName string
Protocol string // "TCP" or "UDP"; never empty
}
// GetServiceEndpoints returns the endpoints for the named service.
// If portName is non-empty, only endpoints matching that port name are returned.
func (c *Client) GetServiceEndpoints(ctx context.Context, serviceName, portName string) ([]Endpoint, error) {
var res api.Endpoints
// TODO: path escape serviceName?
if err := c.do(ctx, "GET", c.nsEndpoint()+"endpoints/"+serviceName, &res); err != nil {
return nil, err
}
var ep []Endpoint
for _, ss := range res.Subsets {
for _, port := range ss.Ports {
if portName != "" && port.Name != portName {
continue
}
for _, addr := range ss.Addresses {
proto := string(port.Protocol)
if proto == "" {
proto = "TCP"
}
ep = append(ep, Endpoint{
IP: addr.IP,
Port: port.Port,
PortName: port.Name,
Protocol: proto,
})
}
}
}
return ep, nil
}
// GetPods returns all pods in the cluster, regardless of status.
func (c *Client) GetPods(ctx context.Context) ([]api.Pod, error) {
var list api.PodList
if err := c.do(ctx, "GET", c.nsEndpoint()+"pods", &list); err != nil {
return nil, err
}
return list.Items, nil
}
// DeletePod deletes the specified Kubernetes pod.
func (c *Client) DeletePod(ctx context.Context, podName string) error {
url := c.nsEndpoint() + "pods/" + podName
req, err := http.NewRequest("DELETE", url, strings.NewReader(`{"gracePeriodSeconds":0}`))
if err != nil {
return fmt.Errorf("failed to create request: DELETE %q : %v", url, err)
}
res, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return fmt.Errorf("failed to make request: DELETE %q: %v", url, err)
}
body, err := io.ReadAll(res.Body)
res.Body.Close()
if err != nil {
return fmt.Errorf("failed to read response body: DELETE %q: %v", url, err)
}
if res.StatusCode != http.StatusOK {
return fmt.Errorf("http error: %d DELETE %q: %q: %v", res.StatusCode, url, string(body), err)
}
return nil
}
// TODO(bradfitz): WatchPod is unreliable, so this is disabled.
//
// AwaitPodNotPending will return a pod's status in a
// podStatusResult when the pod is no longer in the pending
// state.
// The podResourceVersion is required to prevent a pod's entire
// history from being retrieved when the watch is initiated.
// If there is an error polling for the pod's status, or if
// ctx.Done is closed, podStatusResult will contain an error.
func (c *Client) _AwaitPodNotPending(ctx context.Context, podName, podResourceVersion string) (*api.Pod, error) {
if podResourceVersion == "" {
return nil, fmt.Errorf("resourceVersion for pod %v must be provided", podName)
}
ctx, cancel := context.WithCancel(ctx)
defer cancel()
podStatusUpdates, err := c._WatchPod(ctx, podName, podResourceVersion)
if err != nil {
return nil, err
}
for {
select {
case <-ctx.Done():
return nil, ctx.Err()
case psr := <-podStatusUpdates:
if psr.Err != nil {
// If the context is done, prefer its error:
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
return nil, psr.Err
}
}
if psr.Pod.Status.Phase != api.PodPending {
return psr.Pod, nil
}
}
}
}
// PodStatusResult wraps an api.PodStatus and error.
type PodStatusResult struct {
Pod *api.Pod
Type string
Err error
}
type watchPodStatus struct {
// The type of watch update contained in the message
Type string `json:"type"`
// Pod details
Object api.Pod `json:"object"`
}
// TODO(bradfitz): WatchPod is unreliable and sometimes hangs forever
// without closing and sometimes ends prematurely, so this API is
// disabled.
//
// WatchPod long-polls the Kubernetes watch API to be notified
// of changes to the specified pod. Changes are sent on the returned
// PodStatusResult channel as they are received.
// The podResourceVersion is required to prevent a pod's entire
// history from being retrieved when the watch is initiated.
// The provided context must be canceled or timed out to stop the watch.
// If any error occurs communicating with the Kubernetes API, the
// error will be sent on the returned PodStatusResult channel and
// it will be closed.
func (c *Client) _WatchPod(ctx context.Context, podName, podResourceVersion string) (<-chan PodStatusResult, error) {
if podResourceVersion == "" {
return nil, fmt.Errorf("resourceVersion for pod %v must be provided", podName)
}
statusChan := make(chan PodStatusResult, 1)
go func() {
defer close(statusChan)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// Make request to Kubernetes API
getURL := c.endpointURL + "/watch/namespaces/" + c.namespace + "/pods/" + podName
req, err := http.NewRequest("GET", getURL, nil)
req.URL.Query().Add("resourceVersion", podResourceVersion)
if err != nil {
statusChan <- PodStatusResult{Err: fmt.Errorf("failed to create request: GET %q : %v", getURL, err)}
return
}
res, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
statusChan <- PodStatusResult{Err: err}
return
}
defer res.Body.Close()
if res.StatusCode != 200 {
statusChan <- PodStatusResult{Err: fmt.Errorf("WatchPod status %v", res.Status)}
return
}
reader := bufio.NewReader(res.Body)
// bufio.Reader.ReadBytes is blocking, so we watch for
// context timeout or cancellation in a goroutine
// and close the response body when see it. The
// response body is also closed via defer when the
// request is made, but closing twice is OK.
go func() {
<-ctx.Done()
res.Body.Close()
}()
const backupPollDuration = 30 * time.Second
backupPoller := time.AfterFunc(backupPollDuration, func() {
log.Printf("kubernetes: backup poller in WatchPod checking on %q", podName)
st, err := c.PodStatus(ctx, podName)
log.Printf("kubernetes: backup poller in WatchPod PodStatus(%q) = %v, %v", podName, st, err)
if err != nil {
// Some error.
cancel()
}
})
defer backupPoller.Stop()
for {
line, err := reader.ReadBytes('\n')
log.Printf("kubernetes WatchPod status line of %q: %q, %v", podName, line, err)
backupPoller.Reset(backupPollDuration)
if err != nil {
statusChan <- PodStatusResult{Err: fmt.Errorf("error reading streaming response body: %v", err)}
return
}
var wps watchPodStatus
if err := json.Unmarshal(line, &wps); err != nil {
statusChan <- PodStatusResult{Err: fmt.Errorf("failed to decode watch pod status: %v", err)}
return
}
statusChan <- PodStatusResult{Pod: &wps.Object, Type: wps.Type}
}
}()
return statusChan, nil
}
// Retrieve the status of a pod synchronously from the Kube
// API server.
func (c *Client) PodStatus(ctx context.Context, podName string) (*api.PodStatus, error) {
getURL := c.nsEndpoint() + "pods/" + podName // TODO: escape podName?
// Make request to Kubernetes API
req, err := http.NewRequest("GET", getURL, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: GET %q : %v", getURL, err)
}
res, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return nil, fmt.Errorf("failed to make request: GET %q: %v", getURL, err)
}
body, err := io.ReadAll(res.Body)
res.Body.Close()
if err != nil {
return nil, fmt.Errorf("failed to read request body for GET %q: %v", getURL, err)
}
if res.StatusCode != http.StatusOK {
return nil, fmt.Errorf("http error %d GET %q: %q: %v", res.StatusCode, getURL, string(body), err)
}
var pod *api.Pod
if err := json.Unmarshal(body, &pod); err != nil {
return nil, fmt.Errorf("failed to decode pod resources: %v", err)
}
return &pod.Status, nil
}
// PodLog retrieves the container log for the first container
// in the pod.
func (c *Client) PodLog(ctx context.Context, podName string) (string, error) {
// TODO(evanbrown): support multiple containers
url := c.nsEndpoint() + "pods/" + podName + "/log" // TODO: escape podName?
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return "", fmt.Errorf("failed to create request: GET %q : %v", url, err)
}
res, err := ctxhttp.Do(ctx, c.httpClient, req)
if err != nil {
return "", fmt.Errorf("failed to make request: GET %q: %v", url, err)
}
body, err := io.ReadAll(res.Body)
res.Body.Close()
if err != nil {
return "", fmt.Errorf("failed to read response body: GET %q: %v", url, err)
}
if res.StatusCode != http.StatusOK {
return "", fmt.Errorf("http error %d GET %q: %q: %v", res.StatusCode, url, string(body), err)
}
return string(body), nil
}
// GetNodes returns the list of nodes that comprise the Kubernetes cluster
func (c *Client) GetNodes(ctx context.Context) ([]api.Node, error) {
var list api.NodeList
if err := c.do(ctx, "GET", c.endpointURL+"/nodes", &list); err != nil {
return nil, err
}
return list.Items, nil
}