api

README

Flink Monitoring API Golang library

Detail doc: https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/rest_api.html

Status: Beta

package main

import (
	"fmt"

	"github.com/flink-go/api"
)

func main() {
	// Your flink server HTTP API
	c, err := api.New("127.0.0.1:8081")
	if err != nil {
		panic(err)
	}

	// get cluster config
	config, err := c.Config()
	if err != nil {
		panic(err)
	}
	fmt.Println(config)
}

More examples in example dir.

Cluster API

• shutdown cluster
• list config

Jar File API

• upload jar file
• list jar files
• delete jar file
• plan jar file
• run jar file

Job API

• job manager config
• job manager metrics
• list all jobs
• stop a job
• job overview
• job detail

checkpoints

• get all checkpoints of a job
• stop a job with a savepoint

TODO:

• vertices
• checkpoints/config
• /jobs/:jobid/checkpoints/details/:checkpointid
• /jobs/:jobid/config
• /jobs/:jobid/exceptions
• /jobs/:jobid/execution-result
• /jobs/:jobid/metrics
• /jobs/:jobid/plan
• /jobs/:jobid/rescaling
• /jobs/:jobid/rescaling/:triggerid
• overview
• /savepoint-disposal
• /taskmanagers

Documentation

Index

• type Client
  • func New(addr string) (*Client, error)
  • func (c *Client) Checkpoints(jobID string) (checkpointsResp, error)
  • func (c *Client) Config() (configResp, error)
  • func (c *Client) DeleteJar(jarid string) error
  • func (c *Client) Jars() (jarsResp, error)
  • func (c *Client) Job(jobID string) (jobResp, error)
  • func (c *Client) JobManagerConfig() ([]kv, error)
  • func (c *Client) JobManagerMetrics(ids []string) ([]metricValue, error)
  • func (c *Client) JobMetrics(opts JobMetricsOpts) (map[string]interface{}, error)
  • func (c *Client) Jobs() (jobsResp, error)
  • func (c *Client) JobsOverview() (OverviewResp, error)
  • func (c *Client) PlanJar(jarid string) (planResp, error)
  • func (c *Client) RunJar(opts RunOpts) (runResp, error)
  • func (c *Client) SavePoints(jobID string, saveDir string, cancleJob bool) (savePointsResp, error)
  • func (c *Client) Shutdown() error
  • func (c *Client) StopJob(jobID string) error
  • func (c *Client) StopJobWithSavepoint(jobID string, saveDir string, drain bool) (stopJobResp, error)
  • func (c *Client) SubmitJob() error
  • func (c *Client) UploadJar(fpath string) (uploadResp, error)
• type JobMetricsOpts
• type JobOverview
• type OverviewResp
• type RunOpts

Types

type Client

type Client struct {
	// Addr reprents flink job manager server address
	Addr string
	// contains filtered or unexported fields
}

Client reprents flink REST API client

func New

func New(addr string) (*Client, error)

New returns a flink client

func (*Client) Checkpoints

func (c *Client) Checkpoints(jobID string) (checkpointsResp, error)

Checkpoints returns checkpointing statistics for a job.

func (*Client) Config

func (c *Client) Config() (configResp, error)

Config returns the configuration of the WebUI

func (*Client) DeleteJar

func (c *Client) DeleteJar(jarid string) error

DeleteJar deletes a jar file

func (*Client) Jars

func (c *Client) Jars() (jarsResp, error)

Jars eturns a list of all jars previously uploaded via '/jars/upload'

func (*Client) Job

func (c *Client) Job(jobID string) (jobResp, error)

Job returns details of a job.

func (*Client) JobManagerConfig

func (c *Client) JobManagerConfig() ([]kv, error)

JobManagerConfig returns the cluster configuration of job manager server.

func (*Client) JobManagerMetrics

func (c *Client) JobManagerMetrics(ids []string) ([]metricValue, error)

JobManagerMetrics provides access to job manager metrics.

func (*Client) JobMetrics

func (c *Client) JobMetrics(opts JobMetricsOpts) (map[string]interface{}, error)

JobMetrics provides access to aggregated job metrics.

func (*Client) Jobs

func (c *Client) Jobs() (jobsResp, error)

Jobs returns an overview over all jobs and their current state.

func (*Client) JobsOverview

func (c *Client) JobsOverview() (OverviewResp, error)

JobsOverview returns an overview over all jobs.

func (*Client) PlanJar

func (c *Client) PlanJar(jarid string) (planResp, error)

PlanJar returns the dataflow plan of a job contained in a jar previously uploaded via '/jars/upload'. Todo: support more args.

func (*Client) RunJar

func (c *Client) RunJar(opts RunOpts) (runResp, error)

RunJar submits a job by running a jar previously uploaded via '/jars/upload'.

func (*Client) SavePoints

func (c *Client) SavePoints(jobID string, saveDir string, cancleJob bool) (savePointsResp, error)

SavePoints triggers a savepoint, and optionally cancels the job afterwards. This async operation would return a 'triggerid' for further query identifier.

func (*Client) Shutdown

func (c *Client) Shutdown() error

Shutdown shutdown the flink cluster

func (*Client) StopJob

func (c *Client) StopJob(jobID string) error

StopJob terminates a job.

func (*Client) StopJobWithSavepoint

func (c *Client) StopJobWithSavepoint(jobID string, saveDir string, drain bool) (stopJobResp, error)

StopJob stops a job with a savepoint. Optionally, it can also emit a MAX_WATERMARK before taking the savepoint to flush out any state waiting for timers to fire. This async operation would return a 'triggerid' for further query identifier.

func (*Client) SubmitJob

func (c *Client) SubmitJob() error

SubmitJob submits a job.

func (*Client) UploadJar

func (c *Client) UploadJar(fpath string) (uploadResp, error)

Upload uploads jar file

type JobMetricsOpts

type JobMetricsOpts struct {
	// Metrics (optional): string values to select
	// specific metrics.
	Metrics []string

	// Agg (optional): list of aggregation modes which
	// should be calculated. Available aggregations are:
	// "min, max, sum, avg".
	Agg []string

	// Jobs (optional): job list of 32-character
	// hexadecimal strings to select specific jobs.
	Jobs []string
}

type JobOverview

type JobOverview struct {
	ID               string `json:"jid"`
	Name             string `json:"name"`
	State            string `json:"state"`
	Start            int64  `json:"start-time"`
	End              int64  `json:"end-time"`
	Duration         int64  `json:"duration"`
	LastModification int64  `json:"last-modification"`
	Tasks            status `json:"tasks"`
}

type OverviewResp

type OverviewResp struct {
	Jobs []JobOverview `json:"jobs"`
}

type RunOpts

type RunOpts struct {
	// JarID: String value that identifies a jar. When
	// uploading the jar a path is returned, where the
	// filename is the ID.
	JarID string

	// AllowNonRestoredState(optional): Boolean value that
	// specifies whether the job submission should be
	// rejected if the savepoint contains state that
	// cannot be mapped back to the job.
	AllowNonRestoredState bool

	// SavepointPath (optional): String value that
	// specifies the path of the savepoint to restore the
	// job from.
	SavepointPath string

	// programArg (optional): list of program arguments.
	ProgramArg []string

	// EntryClass (optional): String value that specifies
	// the fully qualified name of the entry point class.
	// Overrides the class defined in the jar file
	// manifest.
	EntryClass string

	// Parallelism (optional): Positive integer value that
	// specifies the desired parallelism for the job.
	Parallelism int
}