wip

internal/abstractions/function/function.go

@@ -207,8 +207,13 @@ _stream:
 				break _stream
 			}
 		case <-keyEventChan:
+			exitCode, err := fs.containerRepo.GetContainerExitCode(containerId)
+			if err != nil {
+				return err
+			}
+
 			result, _ := fs.rdb.Get(stream.Context(), Keys.FunctionResult(workspaceName, taskId)).Bytes()
-			if err := stream.Send(&pb.FunctionInvokeResponse{TaskId: taskId, Done: true, Result: result, ExitCode: 0}); err != nil {
+			if err := stream.Send(&pb.FunctionInvokeResponse{TaskId: taskId, Done: true, Result: result, ExitCode: int32(exitCode)}); err != nil {
 				break
 			}
 		case <-ctx.Done():

internal/abstractions/function/function.proto

@@ -22,7 +22,7 @@ message FunctionInvokeResponse {
   string task_id = 1;
   string output = 2;
   bool done = 3;
-  uint32 exit_code = 4;
+  int32 exit_code = 4;
   bytes result = 5;
 }
 

internal/repository/base.go

@@ -29,6 +29,7 @@ type ContainerRepository interface {
 	GetContainerState(string) (*types.ContainerState, error)
 	SetContainerState(string, *types.ContainerState) error
 	SetContainerExitCode(string, int) error
+	GetContainerExitCode(string) (int, error)
 	SetContainerAddress(containerId string, addr string) error
 	UpdateContainerStatus(string, types.ContainerStatus, time.Duration) error
 	DeleteContainerState(*types.ContainerRequest) error

internal/repository/container_redis.go

@@ -85,6 +85,16 @@ func (cr *ContainerRedisRepository) SetContainerExitCode(containerId string, exi
 	return nil
 }
 
+func (cr *ContainerRedisRepository) GetContainerExitCode(containerId string) (int, error) {
+	exitCodeKey := common.RedisKeys.SchedulerContainerExitCode(containerId)
+	exitCode, err := cr.rdb.Get(context.TODO(), exitCodeKey).Int()
+	if err != nil {
+		return -1, err
+	}
+
+	return exitCode, nil
+}
+
 func (cr *ContainerRedisRepository) UpdateContainerStatus(containerId string, status types.ContainerStatus, expiry time.Duration) error {
 	switch status {
 	case types.ContainerStatusPending, types.ContainerStatusRunning, types.ContainerStatusStopping:

internal/worker/image.go

@@ -308,7 +308,6 @@ func (i *ImageClient) Archive(ctx context.Context, bundlePath string, imageId st
 
 	if err != nil {
 		log.Printf("Unable to create archive: %v\n", err)
-		// outputChan <- common.OutputMsg{Done: true, Success: false, Msg: "Unable to archive image."}
 		return err
 	}
 	log.Printf("Container <%v> archive took %v\n", imageId, time.Since(startTime))
@@ -317,11 +316,10 @@ func (i *ImageClient) Archive(ctx context.Context, bundlePath string, imageId st
 	startTime = time.Now()
 	err = i.registry.Push(ctx, archivePath, imageId)
 	if err != nil {
-		log.Printf("Failed to push image for image <%v>: %v\n", imageId, err)
-		// outputChan <- common.OutputMsg{Done: true, Success: false, Msg: "Unable to push image."}
+		log.Printf("Failed to push image <%v>: %v\n", imageId, err)
 		return err
 	}
 
-	log.Printf("Container <%v> push took %v\n", imageId, time.Since(startTime))
+	log.Printf("Image <%v> push took %v\n", imageId, time.Since(startTime))
 	return nil
 }