when subscriber can not get argo pod logs, make pod log message intuitive

chaoscenter/graphql/definitions/shared/chaos_experiment_run.graphqls

@@ -2,7 +2,7 @@ extend type Query {
   """
   Returns experiment run based on experiment run ID
   """
-  getExperimentRun(projectID: ID!, experimentRunID: ID,   notifyID: ID): ExperimentRun!
+  getExperimentRun(projectID: ID!, experimentRunID: ID,   notifyID: ID, infraID: InfraIdentity): ExperimentRun!
 
   """
   Returns the list of experiment run based on various filter parameters

chaoscenter/graphql/definitions/shared/chaos_infrastructure.graphqls

@@ -288,6 +288,10 @@ input PodLogRequest {
   """
   infraID: ID!
   """
+
+  """
+  projectID: ID!
+  """
   ID of a experiment run
   """
   experimentRunID: ID!

chaoscenter/subscriber/pkg/graphql/definations.go

@@ -1,8 +1,14 @@
 package graphql
 
+import (
+	"subscriber/pkg/types"
+)
+
 type SubscriberGql interface {
 	SendRequest(server string, payload []byte) (string, error)
 	MarshalGQLData(gqlData interface{}) (string, error)
+	SendExperimentRunRuquest(infraData map[string]string, podLog types.PodLogRequest) (types.ExperimentRunResponse, error)
+	GenerateExperimentRunPayload(cid, accessKey, version string, podLog types.PodLogRequest) ([]byte, error)
 }
 
 type subscriberGql struct {

chaoscenter/subscriber/pkg/graphql/operations.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"subscriber/pkg/types"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -48,3 +49,32 @@ func (gql *subscriberGql) MarshalGQLData(gqlData interface{}) (string, error) {
 	processed = strings.Replace(processed, `\"`, `\\\"`, -1)
 	return processed, nil
 }
+
+// Get an experiment run on the GraphQL server
+func (gql *subscriberGql) SendExperimentRunRuquest(infraData map[string]string, podLog types.PodLogRequest) (types.ExperimentRunResponse, error) {
+
+	payload, _ := gql.GenerateExperimentRunPayload(infraData["INFRA_ID"], infraData["ACCESS_KEY"], infraData["VERSION"], podLog)
+
+	body, err := gql.SendRequest(infraData["SERVER_ADDR"], payload)
+	if err != nil {
+		log.WithError(err).Print("Failed to send experiment run request")
+	}
+
+	var respsone types.ExperimentRunResponse
+	err = json.Unmarshal([]byte(body), &respsone)
+	if err != nil {
+		log.WithError(err).WithField("data", string(body)).Fatal("Failed to parse ExperimentRun data")
+	}
+
+	log.Print("Response from the server: ", body)
+
+	return respsone, nil
+}
+
+func (gql *subscriberGql) GenerateExperimentRunPayload(cid, accessKey, version string, podLog types.PodLogRequest) ([]byte, error) {
+	infraID := `infraID: {infraID: \"` + cid + `\", version: \"` + version + `\", accessKey: \"` + accessKey + `\"}`
+	query := infraID + `, experimentRunID: \"` + podLog.ExperimentRunID + `\", projectID: \"` + podLog.ProjectID + `\", notifyID: \"\"`
+
+	var payload = []byte(`{"query": "query { getExperimentRun(` + query + `) { phase } }"}`)
+	return payload, nil
+}

chaoscenter/subscriber/pkg/k8s/defination.go

@@ -15,9 +15,11 @@ import (
 
 type SubscriberK8s interface {
 	GetLogs(podName, namespace, container string) (string, error)
-	CreatePodLog(podLog types.PodLogRequest) (types.PodLog, error)
+	GetPodLogs(infraData map[string]string, podLog types.PodLogRequest) (types.PodLog, error)
+	categorizeLogByPhase(phase string) string
+	CreatePodLog(infraData map[string]string, logDetails types.PodLog, podLog types.PodLogRequest) (types.PodLog, error)
 	SendPodLogs(infraData map[string]string, podLog types.PodLogRequest)
-	GenerateLogPayload(cid, accessKey, version string, podLog types.PodLogRequest) ([]byte, error)
+	GenerateLogPayload(cid, accessKey, version, processed string, podLog types.PodLogRequest) ([]byte, error)
 	GetKubernetesNamespaces(request types.KubeNamespaceRequest) ([]*types.KubeNamespace, error)
 	GetKubernetesObjects(request types.KubeObjRequest) (*types.KubeObject, error)
 	GetObjectDataByNamespace(namespace string, dynamicClient dynamic.Interface, resourceType schema.GroupVersionResource) ([]types.ObjectData, error)

chaoscenter/subscriber/pkg/k8s/log.go

@@ -53,17 +53,7 @@ func (k8s *k8sSubscriber) GetLogs(podName, namespace, container string) (string,
 }
 
 // create pod log for normal pods and chaos-engine pods
-func (k8s *k8sSubscriber) CreatePodLog(podLog types.PodLogRequest) (types.PodLog, error) {
-	logDetails := types.PodLog{}
-	mainLog, err := k8s.GetLogs(podLog.PodName, podLog.PodNamespace, "main")
-	// try getting argo pod logs
-	if err != nil {
-		logrus.Errorf("Failed to get argo pod %v logs, err: %v", podLog.PodName, err)
-		logDetails.MainPod = "Failed to get argo pod logs"
-	} else {
-		logDetails.MainPod = strconv.Quote(strings.Replace(mainLog, `"`, `'`, -1))
-		logDetails.MainPod = logDetails.MainPod[1 : len(logDetails.MainPod)-1]
-	}
+func (k8s *k8sSubscriber) CreatePodLog(infraData map[string]string, logDetails types.PodLog, podLog types.PodLogRequest) (types.PodLog, error) {
 	// try getting experiment pod logs if requested
 	if strings.ToLower(podLog.PodType) == "chaosengine" && podLog.ChaosNamespace != nil {
 		chaosLog := make(map[string]string)
@@ -96,32 +86,84 @@ func (k8s *k8sSubscriber) CreatePodLog(podLog types.PodLogRequest) (types.PodLog
 
 // SendPodLogs generates graphql mutation to send events updates to graphql server
 func (k8s *k8sSubscriber) SendPodLogs(infraData map[string]string, podLog types.PodLogRequest) {
-	// generate graphql payload
-	payload, err := k8s.GenerateLogPayload(infraData["INFRA_ID"], infraData["ACCESS_KEY"], infraData["VERSION"], podLog)
+
+	logDetails, err := k8s.GetPodLogs(infraData, podLog)
 	if err != nil {
-		logrus.WithError(err).Print("Error while retrieving the workflow logs")
+		logrus.WithError(err).Print("failed to get pod logs")
 	}
+
+	// MarashalGQLData
+	processed, err := k8s.gqlSubscriberServer.MarshalGQLData(logDetails)
+	if err != nil {
+		processed = " Could not get logs "
+	}
+
+	// Generate graphql payload
+	payload, _ := k8s.GenerateLogPayload(infraData["INFRA_ID"], infraData["ACCESS_KEY"], infraData["VERSION"], processed, podLog)
+
 	body, err := k8s.gqlSubscriberServer.SendRequest(infraData["SERVER_ADDR"], payload)
 	if err != nil {
 		logrus.Print(err.Error())
 	}
 	logrus.Print("Response from the server: ", body)
 }
 
-func (k8s *k8sSubscriber) GenerateLogPayload(cid, accessKey, version string, podLog types.PodLogRequest) ([]byte, error) {
-	infraID := `{infraID: \"` + cid + `\", version: \"` + version + `\", accessKey: \"` + accessKey + `\"}`
-	processed := " Could not get logs "
+func (k8s *k8sSubscriber) GetPodLogs(infraData map[string]string, podLog types.PodLogRequest) (types.PodLog, error) {
+	logDetails := types.PodLog{}
+	mainLog, err := k8s.GetLogs(podLog.PodName, podLog.PodNamespace, "main")
+
+	// try getting argo pod logs
+	if err != nil {
+
+		// fetch ExperimentRun Phase from graphql
+		experimentRun, _ := k8s.gqlSubscriberServer.SendExperimentRunRuquest(infraData, podLog)
+		logDetails.MainPod = k8s.categorizeLogByPhase(experimentRun.Data.ExperimentRun.Phase) + " Pod name: " + podLog.PodName
+
+		logrus.WithError(err).Print("Failed to get pod logs for attempting to fetch experiment run phase")
+
+	} else {
+
+		logDetails.MainPod = strconv.Quote(strings.Replace(mainLog, `"`, `'`, -1))
+		logDetails.MainPod = logDetails.MainPod[1 : len(logDetails.MainPod)-1]
+
+		logDetails, err = k8s.CreatePodLog(infraData, logDetails, podLog)
 
-	// get the logs
-	logDetails, err := k8s.CreatePodLog(podLog)
-	if err == nil {
-		// process log data
-		processed, err = k8s.gqlSubscriberServer.MarshalGQLData(logDetails)
 		if err != nil {
-			processed = " Could not get logs "
+			return logDetails, err
 		}
 	}
 
+	return logDetails, nil
+}
+
+// Categorizing log messages by experiment phase
+func (k8s *k8sSubscriber) categorizeLogByPhase(phase string) string {
+	switch phase {
+	case "Completed":
+		return "Experiment pod was deleted."
+	case "Stopped":
+		return "Experiment is stopped."
+	case "Running":
+		return "Experiment pod is initializing."
+	case "Queue":
+		return "Queue."
+	case "NA":
+		return "NA."
+	case "Terminated":
+		return "Terminated."
+	case "Completed_With_Error":
+		return "Completed_With_Error."
+	case "Timeout":
+		return "Timeout."
+	case "Error":
+		return "Experiment could not start."
+	default:
+		return "Experiment pod is initializing."
+	}
+}
+
+func (k8s *k8sSubscriber) GenerateLogPayload(cid, accessKey, version, processed string, podLog types.PodLogRequest) ([]byte, error) {
+	infraID := `{infraID: \"` + cid + `\", version: \"` + version + `\", accessKey: \"` + accessKey + `\"}`
 	mutation := `{ infraID: ` + infraID + `, requestID:\"` + podLog.RequestID + `\", experimentRunID: \"` + podLog.ExperimentRunID + `\", podName: \"` + podLog.PodName + `\", podType: \"` + podLog.PodType + `\", log:\"` + processed[1:len(processed)-1] + `\"}`
 	var payload = []byte(`{"query":"mutation { podLog(request:` + mutation + ` )}"}`)
 

chaoscenter/subscriber/pkg/types/graphql.go

@@ -0,0 +1,13 @@
+package types
+
+type ExperimentRunResponse struct {
+	Data ExperimentRun `json:"data"`
+}
+
+type ExperimentRun struct {
+	ExperimentRun Phase `json:"getExperimentRun"`
+}
+
+type Phase struct {
+	Phase string `json:"phase"`
+}

chaoscenter/subscriber/pkg/types/log.go

@@ -2,6 +2,7 @@ package types
 
 type PodLogRequest struct {
 	RequestID       string  `json:"requestID"`
+	ProjectID       string  `json:"projectID"`
 	InfraID         string  `json:"infraID"`
 	ExperimentRunID string  `json:"experimentRunID"`
 	PodName         string  `json:"podName"`