Merge pull request #8 from ilikeorangutans/check_project

Check project

Author: ilikeorangutans

azkaban/client.go

@@ -86,13 +86,13 @@ func (c *Client) FetchExecutionJobLog(executionID int64, jobID string, offset in
 	return log, err
 }
 
-func (c *Client) FlowExecutions(project, flow string) (Executions, error) {
+func (c *Client) FlowExecutions(project, flow string, paginator Paginator) (Executions, error) {
 	params := make(map[string]string)
 	params["ajax"] = "fetchFlowExecutions"
 	params["project"] = project
 	params["flow"] = flow
-	params["start"] = "0"
-	params["length"] = "20"
+	params["start"] = strconv.Itoa(paginator.Offset)
+	params["length"] = strconv.Itoa(paginator.Length)
 
 	executions := ExecutionsList{}
 	if err := c.requestAndDecode("GET", "manager", params, &executions); err != nil {
@@ -257,15 +257,6 @@ func findExecutions(n *htmlx.Node) ([]FlowExecution, error) {
 	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 {
@@ -281,7 +272,7 @@ func findElementsOfType(n *htmlx.Node, t string) []*htmlx.Node {
 
 func findElementWithID(n *htmlx.Node, id string) *htmlx.Node {
 	if hasAttribute(n, "id", id) {
-		return n;
+		return n
 	}
 	for c := n.FirstChild; c != nil; c = c.NextSibling {
 		if result := findElementWithID(c, id); result != nil {

azkaban/executions.go

@@ -15,5 +15,5 @@ type ajaxExecRepo struct {
 }
 
 func (r *ajaxExecRepo) ListExecutions(proj Project, f Flow, p Paginator) (Executions, error) {
-	return r.azkaban.FlowExecutions(proj.Name, f.FlowID)
+	return r.azkaban.FlowExecutions(proj.Name, f.FlowID, p)
 }

azkaban/flow_predicates.go

@@ -0,0 +1,42 @@
+package azkaban
+
+import (
+	"log"
+	"regexp"
+	"strings"
+)
+
+type FlowPredicate func(Flow) bool
+
+func MatchesAnyFlow() FlowPredicate {
+	return func(_ Flow) bool { return true }
+}
+func MatchesFlowName(input string) FlowPredicate {
+	if len(input) > 0 {
+		regexString := input
+		if strings.HasPrefix(regexString, "/") && strings.HasSuffix(regexString, "/") {
+			regex, err := regexp.Compile(regexString[1 : len(input)-1])
+			if err != nil {
+				log.Fatal(err)
+			}
+			return func(f Flow) bool { return regex.MatchString(f.FlowID) }
+		} else {
+			return func(f Flow) bool { return strings.HasPrefix(f.FlowID, input) }
+		}
+	} else {
+		return func(f Flow) bool { return true }
+	}
+
+}
+
+// MatchesAll returns a FlowPredicate that returns true if a flow matches all the provided predicates.
+func MatchesAll(predicates ...FlowPredicate) FlowPredicate {
+	return func(f Flow) bool {
+		for _, p := range predicates {
+			if !p(f) {
+				return false
+			}
+		}
+		return true
+	}
+}

azkaban/flows.go

@@ -1,9 +1,17 @@
 package azkaban
 
+import "sort"
+
+type Flows []Flow
+
+func (f Flows) Len() int           { return len(f) }
+func (f Flows) Swap(i, j int)      { f[i], f[j] = f[j], f[i] }
+func (f Flows) Less(i, j int) bool { return f[i].FlowID < f[j].FlowID }
+
 type FlowRepository interface {
 	// Flow returns a flow instance for the given name and project
 	Flow(Project, string) (Flow, Project, error)
-	ListFlows(Project) ([]Flow, error)
+	ListFlows(Project, FlowPredicate) (Flows, error)
 }
 
 func NewFlowRepository(client *Client) FlowRepository {
@@ -16,8 +24,22 @@ type ajaxFlowRepoImpl struct {
 	azkaban *Client
 }
 
-func (r *ajaxFlowRepoImpl) ListFlows(proj Project) ([]Flow, error) {
-	return r.azkaban.ListFlows(proj.Name)
+func (r *ajaxFlowRepoImpl) ListFlows(proj Project, predicate FlowPredicate) (Flows, error) {
+	flows, err := r.azkaban.ListFlows(proj.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	var result Flows
+	for _, flow := range flows {
+		if predicate(flow) {
+			result = append(result, flow)
+		}
+	}
+
+	sort.Sort(result)
+
+	return result, nil
 }
 
 func (r *ajaxFlowRepoImpl) Flow(proj Project, flowID string) (Flow, Project, error) {

azkaban/health.go

@@ -18,6 +18,15 @@ func (h Health) Colored() string {
 	}
 }
 
+func (h Health) IsHealthy() bool {
+	switch h {
+	case Healthy:
+		return true
+	default:
+		return false
+	}
+}
+
 const (
 	Healthy    Health = "healthy"
 	Concerning        = "concerning"

azkaban/paginator.go

@@ -5,4 +5,11 @@ type Paginator struct {
 	Offset int
 }
 
+func NMostRecent(n int) Paginator {
+	return Paginator{
+		Length: n,
+		Offset: 0,
+	}
+}
+
 var TenMostRecent = Paginator{Length: 10, Offset: 0}

azkaban/projects.go

@@ -1,7 +1,13 @@
 package azkaban
 
+import (
+	"errors"
+	"fmt"
+)
+
 type ProjectRepository interface {
 	ListProjects() ([]Project, error)
+	ByName(name string) (Project, error)
 }
 
 func NewProjectRepository(client *Client) ProjectRepository {
@@ -14,6 +20,20 @@ type projectRepoImpl struct {
 	client *Client
 }
 
+func (r *projectRepoImpl) ByName(name string) (Project, error) {
+	projects, err := r.ListProjects()
+	if err != nil {
+		return Project{}, err
+	}
+
+	for _, p := range projects {
+		if p.Name == name {
+			return p, nil
+		}
+	}
+	return Project{}, errors.New(fmt.Sprintf("no project with name %s", name))
+}
+
 func (r *projectRepoImpl) ListProjects() ([]Project, error) {
 	return r.client.ListProjects()
 }

azkaban/status.go

@@ -9,7 +9,7 @@ import (
 type Status string
 
 func (s Status) Colored() string {
-	return s.ColorFunc()((fmt.Sprintf("%-9s", s)))
+	return s.ColorFunc()(fmt.Sprintf("%-9s", s))
 }
 
 func (s Status) ColorFunc() func(string, ...interface{}) string {

cli/check_cmd.go

@@ -0,0 +1,18 @@
+package cli
+
+import (
+	"github.com/spf13/cobra"
+)
+
+func NewCheckCmd(context Context) *cobra.Command {
+	checkCmd := &cobra.Command{
+		Use:     "check",
+		Aliases: []string{"c"},
+		Short:   "checks things",
+	}
+
+	checkCmd.AddCommand(newCheckFlowCmd(context))
+	checkCmd.AddCommand(newCheckProjectCmd(context))
+
+	return checkCmd
+}

cli/check_flow_cmd.go

@@ -12,14 +12,8 @@ import (
 	"strings"
 )
 
-func NewCheckFlowCmd(context Context) *cobra.Command {
-	checkCmd := &cobra.Command{
-		Use:     "check",
-		Aliases: []string{"c"},
-		Short:   "checks things",
-	}
-
-	checkFlowCmd := &cobra.Command{
+func newCheckFlowCmd(context Context) *cobra.Command {
+	return &cobra.Command{
 		Use:     "flow",
 		Aliases: []string{"f"},
 		Short:   "check a given flow",
@@ -110,9 +104,6 @@ func NewCheckFlowCmd(context Context) *cobra.Command {
 			}
 		},
 	}
-	checkCmd.AddCommand(checkFlowCmd)
-
-	return checkCmd
 }
 
 type FlowStatusChecker struct {
@@ -140,7 +131,7 @@ func (h FlowStatusChecker) printSchedule() error {
 func (h FlowStatusChecker) printFlowStatus() (status FlowStatus, err error) {
 	fmt.Printf("Checking status of %s %s...\n", h.project.Name, h.flow.FlowID)
 	client := h.client
-	executions, err := client.FlowExecutions(h.project.Name, h.flow.FlowID)
+	executions, err := client.FlowExecutions(h.project.Name, h.flow.FlowID, azkaban.TenMostRecent)
 	if err != nil {
 		return status, err
 	}