work in progress, fetching executions via html now

Author: ilikeorangutans

azkaban/client.go

@@ -2,13 +2,17 @@ package azkaban
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
+	htmlx "golang.org/x/net/html"
 	"html"
 	"io"
 	"io/ioutil"
 	"log"
 	"net/http"
 	"net/http/httputil"
+	"strconv"
+	"time"
 )
 
 type Client struct {
@@ -129,7 +133,7 @@ func (c *Client) RestartFlowNow(project, flow string) error {
 	return nil
 }
 
-func (c *Client) FlowEcecutionStatus(executionID int64) (FlowExecutionStatus, error) {
+func (c *Client) FlowExecutionStatus(executionID int64) (FlowExecutionStatus, error) {
 	status := FlowExecutionStatus{}
 
 	params := make(map[string]string)
@@ -182,6 +186,120 @@ func (c *Client) FlowSchedule(projectID int64, flowID string) (FlowSchedule, err
 	}
 }
 
+func (c *Client) Running() ([]FlowExecution, error) {
+	resp, err := c.request("GET", "executor", nil)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != 200 {
+		return nil, errors.New(fmt.Sprintf("got %d %s when retrieving running executions", resp.StatusCode, resp.Status))
+	}
+
+	defer resp.Body.Close()
+	doc, err := htmlx.Parse(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	// Azkaban serves the login page simply with a HTTP 200 so the only way to check if we're looking at the login page
+	// is by looking for the login element.
+	if findElementWithID(doc,"username") != nil && findElementWithID(doc, "password") != nil {
+		return nil, errors.New("credentials expired, reauthenticate")
+	}
+
+	table := findElementWithID(doc, "executingJobs")
+
+	return findExecutions(table)
+}
+
+type FlowExecution struct {
+	FlowID  string
+	Project string
+	Execution
+}
+
+func findExecutions(n *htmlx.Node) ([]FlowExecution, error) {
+	tbody := findElementsOfType(n, "tbody")
+	rows := findElementsOfType(tbody[0], "tr")
+
+	var executions []FlowExecution
+	for _, row := range rows {
+		cells := findElementsOfType(row, "td")
+
+		execID, err := strconv.ParseInt(findElementsOfType(cells[1], "a")[0].FirstChild.Data, 10, 64)
+		if err != nil {
+			return nil, err
+		}
+		flowID := findElementsOfType(cells[3], "a")[0].FirstChild.Data
+		project := findElementsOfType(cells[4], "a")[0].FirstChild.Data
+		// TODO project ID is not part of the actual link so we can't parse it. We'll have to fetch projects before we do this
+		// parsing time "2019-08-21 01:53:42" as "Jan 2, 2006 at 3:04pm": cannot parse "2019-08-21 01:53:42" as "Jan"
+		startTime, err := time.Parse("2006-01-02 15:04:05", cells[7].FirstChild.Data)
+		if err != nil {
+			return nil, err
+		}
+
+		execution := FlowExecution{
+			FlowID:  flowID,
+			Project: project,
+			Execution: Execution{
+				SubmitTime: AzkabanTimestamp{},
+				StartTime:  AzkabanTimestamp(startTime),
+				Status:     "RUNNING",
+				ID:         execID,
+				EndTime:    AzkabanTimestamp(time.Now()),
+			},
+		}
+
+		executions = append(executions, execution)
+	}
+
+	return executions, nil
+}
+
+func getAttribute(n *htmlx.Node, name string) string {
+	for _, a := range n.Attr {
+		if a.Key == name {
+			return a.Val
+		}
+	}
+	return ""
+}
+
+func findElementsOfType(n *htmlx.Node, t string) []*htmlx.Node {
+	var result []*htmlx.Node
+	if n.Type == htmlx.ElementNode && n.Data == t {
+		result = append(result, n)
+	}
+
+	for c := n.FirstChild; c != nil; c = c.NextSibling {
+		result = append(result, findElementsOfType(c, t)...)
+	}
+
+	return result
+}
+
+func findElementWithID(n *htmlx.Node, id string) *htmlx.Node {
+	if hasAttribute(n, "id", id) {
+		return n;
+	}
+	for c := n.FirstChild; c != nil; c = c.NextSibling {
+		if result := findElementWithID(c, id); result != nil {
+			return result
+		}
+	}
+	return nil
+}
+
+func hasAttribute(n *htmlx.Node, name string, value string) bool {
+	for _, a := range n.Attr {
+		if a.Key == name && a.Val == value {
+			return true
+		}
+	}
+	return false
+}
+
 func (c *Client) requestAndDecode(method string, path string, params map[string]string, dst interface{}) error {
 	resp, err := c.request(method, path, params)
 	if err != nil {

azkaban/json.go

@@ -258,22 +258,6 @@ func (t AzkabanStringTime) Time() time.Time {
 	return time.Time(t)
 }
 
-func formatDuration(d time.Duration) string {
-	hours := 0
-	if d.Hours() >= 1.0 {
-		hours = int(d.Hours())
-	}
-	minutes := 0
-	if d.Minutes() > 0 {
-		minutes = int(d.Minutes()) % 60
-	}
-	seconds := 0
-	if d.Seconds() > 0 {
-		seconds = int(d.Seconds()) % 60
-	}
-	return fmt.Sprintf("%d:%02d:%02d", hours, minutes, seconds)
-}
-
 type FlowExecutionStatus struct {
 	Attempt   int         `json:"attempt"`
 	Status    Status      `json:"status"`

check_flow_action.go

@@ -207,7 +207,7 @@ func (h FlowStatusChecker) printFlowStatus() (status FlowStatus, err error) {
 		status.LastExecution = executions.MostRecentExecution()
 		executionID := executions.MostRecentExecution().ID
 
-		flowExecStatus, err := client.FlowEcecutionStatus(executionID)
+		flowExecStatus, err := client.FlowExecutionStatus(executionID)
 		if err != nil {
 			return status, err
 		}

get_action.go

@@ -40,6 +40,12 @@ func SetupGetActions() cli.Command {
 				Action:    handlers.GetExecutionsAction,
 				ArgsUsage: "flow",
 			},
+			{
+				Name:    "running",
+				Aliases: []string{"r"},
+				Usage:   "lists running executions",
+				Action:  handlers.GetRunningAction,
+			},
 		},
 	}
 }
@@ -48,6 +54,37 @@ type GetActionsHandler struct {
 	ActionWithContext
 }
 
+func (a *GetActionsHandler) GetRunningAction(c *cli.Context) error {
+	executions, err := a.Client().Running()
+
+	w := new(tabwriter.Writer)
+	w.Init(os.Stdout, 4, 4, 2, ' ', 0)
+	fmt.Fprintf(
+		w,
+		"%s\t%s\t%s\t%s\t%s\n",
+		"Project",
+		"FlowID",
+		color.WhiteString("Status"),
+		"Start time",
+		"Runtime",
+	)
+
+	for _, execution := range executions {
+		fmt.Fprintf(
+			w,
+			"%s\t%s\t%s\t%s\t%s\n",
+			execution.Project,
+			execution.FlowID,
+			execution.Status.Colored(),
+			humanize.Time(execution.StartTime.Time()),
+			format.DurationHumanReadable(time.Since(execution.StartTime.Time())),
+		)
+	}
+
+	w.Flush()
+	return err
+}
+
 func (a *GetActionsHandler) GetExecutionsAction(c *cli.Context) error {
 	if !c.Args().Present() {
 		return errors.New("no flowID given")

go.mod

@@ -8,5 +8,5 @@ require (
 	github.com/mattn/go-colorable v0.0.9 // indirect
 	github.com/mattn/go-isatty v0.0.3 // indirect
 	github.com/urfave/cli v1.21.0
-	golang.org/x/sys v0.0.0-20171110072704-1e2299c37cc9 // indirect
+	golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7
 )

go.sum

@@ -9,8 +9,13 @@ github.com/mattn/go-isatty v0.0.3 h1:ns/ykhmWi7G9O+8a448SecJU3nSMBXJfqQkl0upE1jI
 github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
 github.com/urfave/cli v1.21.0 h1:wYSSj06510qPIzGSua9ZqsncMmWE3Zr55KBERygyrxE=
 github.com/urfave/cli v1.21.0/go.mod h1:lxDj6qX9Q6lWQxIrbrT0nwecwUtRnhVZAJjJZrVUZZQ=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7 h1:fHDIZ2oxGnUZRN6WgWFCbYBjH9uqVPRCUVUDhs0wnbA=
+golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sys v0.0.0-20171110072704-1e2299c37cc9 h1:LRRcaqNAXmlmNmT+Q2u9z+iEf5+Y+S1Ea9MNwGxOc98=
 golang.org/x/sys v0.0.0-20171110072704-1e2299c37cc9/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=