fix: handle empty metadata->resource in job spec & added grpc logger … (

#196)

* fix: handle empty metadata->resource in job spec & added grpc logger for stream interceptor

* fix: added same situation in test case

core/job/handler/v1beta1/job_adapter.go

@@ -372,8 +372,14 @@ func toMetadata(jobMetadata *pb.JobMetadata) (*job.Metadata, error) {
 
 	if jobMetadata.Resource != nil {
 		metadataResourceProto := jobMetadata.Resource
-		request := job.NewMetadataResourceConfig(metadataResourceProto.Request.Cpu, metadataResourceProto.Request.Memory)
-		limit := job.NewMetadataResourceConfig(metadataResourceProto.Limit.Cpu, metadataResourceProto.Limit.Memory)
+		var request *job.MetadataResourceConfig
+		if metadataResourceProto.Request != nil {
+			request = job.NewMetadataResourceConfig(metadataResourceProto.Request.Cpu, metadataResourceProto.Request.Memory)
+		}
+		var limit *job.MetadataResourceConfig
+		if metadataResourceProto.Limit != nil {
+			limit = job.NewMetadataResourceConfig(metadataResourceProto.Limit.Cpu, metadataResourceProto.Limit.Memory)
+		}
 		resourceMetadata := job.NewResourceMetadata(request, limit)
 		metadataBuilder = metadataBuilder.WithResource(resourceMetadata)
 	}

core/job/handler/v1beta1/job_test.go

@@ -916,6 +916,23 @@ func TestNewJobHandler(t *testing.T) {
 					WindowSize:       jobWindow.GetSize(),
 					WindowOffset:     jobWindow.GetOffset(),
 					WindowTruncateTo: jobWindow.GetTruncateTo(),
+					Config:           []*pb.JobConfigItem{},
+					Window:           &pb.JobSpecification_Window{},
+					Dependencies:     []*pb.JobDependency{},
+					Assets:           map[string]string{},
+					Hooks:            []*pb.JobSpecHook{},
+					Description:      "",
+					Labels:           map[string]string{},
+					Behavior:         &pb.JobSpecification_Behavior{Notify: []*pb.JobSpecification_Behavior_Notifiers{}},
+					Metadata: &pb.JobMetadata{
+						Resource: &pb.JobSpecMetadataResource{
+							Request: nil,
+							Limit:   nil,
+						},
+						Airflow: &pb.JobSpecMetadataAirflow{},
+					},
+					Destination: "",
+					Sources:     []string{},
 				},
 				{
 					Version:          int32(jobVersion),

server/server.go

@@ -88,6 +88,7 @@ func setupGRPCServer(l log.Logger) (*grpc.Server, error) {
 		),
 		grpc_middleware.WithStreamServerChain(
 			otelgrpc.StreamServerInterceptor(),
+			grpc_logrus.StreamServerInterceptor(grpcLogrusEntry, opts...),
 			grpc_prometheus.StreamServerInterceptor,
 			grpc_recovery.StreamServerInterceptor(grpc_recovery.WithRecoveryHandler(recoverPanic)),
 		),