Merge pull request #113 from astronomerio/feature/88-output-logs-for-…

…local-development

Add command to output local airflow logs

airflow/docker.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"text/tabwriter"
 
+	"github.com/docker/libcompose/cli/logger"
 	dockercompose "github.com/docker/libcompose/docker"
 	"github.com/docker/libcompose/docker/ctx"
 	"github.com/docker/libcompose/project"
@@ -115,8 +116,9 @@ func createProject(projectName, airflowHome string) (project.APIProject, error)
 	// Create the project
 	project, err := dockercompose.NewProject(&ctx.Context{
 		Context: project.Context{
-			ComposeBytes: [][]byte{[]byte(yaml)},
-			ProjectName:  projectName,
+			ComposeBytes:  [][]byte{[]byte(yaml)},
+			ProjectName:   projectName,
+			LoggerFactory: logger.NewColorLoggerFactory(),
 		},
 	}, nil)
 
@@ -188,6 +190,44 @@ func Kill(airflowHome string) error {
 	return nil
 }
 
+// Logs out airflow webserver or scheduler logs
+func Logs(airflowHome string, webserver, scheduler, follow bool) error {
+	s := make([]string, 0)
+
+	// Get project name from config
+	projectName := config.CFG.ProjectName.GetString()
+
+	// Create libcompose project
+	project, err := createProject(projectName, airflowHome)
+	if err != nil {
+		return errors.Wrap(err, messages.COMPOSE_CREATE_ERROR)
+	}
+
+	psInfo, err := project.Ps(context.Background())
+	if err != nil {
+		return errors.Wrap(err, messages.COMPOSE_STATUS_CHECK_ERROR)
+	}
+
+	if len(psInfo) == 0 {
+		fmt.Println("Project is not running, first start the project to view logs")
+		os.Exit(1)
+	}
+
+	if scheduler {
+		s = append(s, "scheduler")
+	}
+	if webserver {
+		s = append(s, "webserver")
+	}
+
+	err = project.Log(context.Background(), follow, s...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // Stop a running docker project
 func Stop(airflowHome string) error {
 	// Get project name from config

cmd/airflow.go

@@ -22,8 +22,11 @@ import (
 )
 
 var (
-	projectName string
-	forceDeploy bool
+	projectName   string
+	followLogs    bool
+	forceDeploy   bool
+	schedulerLogs bool
+	webserverLogs bool
 
 	airflowRootCmd = &cobra.Command{
 		Use:   "airflow",
@@ -63,6 +66,14 @@ var (
 		RunE:   airflowKill,
 	}
 
+	airflowLogsCmd = &cobra.Command{
+		Use:    "logs",
+		Short:  "Output logs for a development airflow cluster",
+		Long:   "Output logs for a development airflow cluster",
+		PreRun: ensureProjectDir,
+		RunE:   airflowLogs,
+	}
+
 	airflowStopCmd = &cobra.Command{
 		Use:    "stop",
 		Short:  "Stop a development airflow cluster",
@@ -99,6 +110,12 @@ func init() {
 	// Airflow kill
 	airflowRootCmd.AddCommand(airflowKillCmd)
 
+	// Airflow logs
+	airflowRootCmd.AddCommand(airflowLogsCmd)
+	airflowLogsCmd.Flags().BoolVarP(&followLogs, "follow", "f", false, "Follow log output")
+	airflowLogsCmd.Flags().BoolVarP(&schedulerLogs, "scheduler", "s", false, "Output scheduler logs")
+	airflowLogsCmd.Flags().BoolVarP(&webserverLogs, "webserver", "w", false, "Output webserver logs")
+
 	// Airflow stop
 	airflowRootCmd.AddCommand(airflowStopCmd)
 
@@ -179,6 +196,17 @@ func airflowKill(cmd *cobra.Command, args []string) error {
 	return airflow.Kill(config.WorkingPath)
 }
 
+// Outputs logs for a development airflow cluster
+func airflowLogs(cmd *cobra.Command, args []string) error {
+	// default is to display all logs
+	if !schedulerLogs && !webserverLogs {
+		schedulerLogs = true
+		webserverLogs = true
+	}
+
+	return airflow.Logs(config.WorkingPath, webserverLogs, schedulerLogs, followLogs)
+}
+
 // Stop an airflow cluster
 func airflowStop(cmd *cobra.Command, args []string) error {
 	return airflow.Stop(config.WorkingPath)