Merge pull request flyteorg#72 from lyft/fairness-hard-cap

Implementing project-level and namespace-level quota cap

flytepropeller/go.mod

@@ -3,8 +3,11 @@ module github.com/lyft/flytepropeller
 go 1.13
 
 require (
+	cloud.google.com/go v0.53.0 // indirect
+	github.com/Azure/azure-sdk-for-go v39.1.0+incompatible // indirect
+	github.com/Azure/go-autorest/autorest v0.9.5 // indirect
 	github.com/DiSiqueira/GoTree v1.0.1-0.20180907134536-53a8e837f295
-	github.com/aws/aws-sdk-go v1.28.12 // indirect
+	github.com/aws/aws-sdk-go v1.29.2 // indirect
 	github.com/benlaurie/objecthash v0.0.0-20180202135721-d1e3d6079fc1
 	github.com/coreos/go-oidc v2.2.1+incompatible // indirect
 	github.com/fatih/color v1.9.0
@@ -14,30 +17,31 @@ require (
 	github.com/golang/protobuf v1.3.3
 	github.com/google/uuid v1.1.1
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.0
+	github.com/grpc-ecosystem/grpc-gateway v1.13.0 // indirect
 	github.com/imdario/mergo v0.3.8 // indirect
 	github.com/lyft/datacatalog v0.2.1
-	github.com/lyft/flyteidl v0.17.2
-	github.com/lyft/flyteplugins v0.3.2
+	github.com/lyft/flyteidl v0.17.7
+	github.com/lyft/flyteplugins v0.3.9
+	//github.com/lyft/flyteplugins v0.3.4-0.20200218200451-1ce137d446ba
 	github.com/lyft/flytestdlib v0.3.2
 	github.com/magiconair/properties v1.8.1
 	github.com/mitchellh/mapstructure v1.1.2
 	github.com/ncw/swift v1.0.50 // indirect
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.4.0
+	github.com/prometheus/client_golang v1.4.1
 	github.com/spf13/cobra v0.0.5
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.4.0
-	golang.org/x/crypto v0.0.0-20200206161412-a0c6ece9d31a // indirect
+	golang.org/x/crypto v0.0.0-20200210222208-86ce3cb69678 // indirect
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
-	google.golang.org/api v0.17.0 // indirect
 	google.golang.org/grpc v1.27.1
 	gopkg.in/ini.v1 v1.52.0 // indirect
-	k8s.io/api v0.17.2
-	k8s.io/apimachinery v0.17.2
+	k8s.io/api v0.17.3
+	k8s.io/apimachinery v0.17.3
 	k8s.io/client-go v11.0.0+incompatible
 	k8s.io/klog v1.0.0
 	k8s.io/kube-openapi v0.0.0-20200204173128-addea2498afe // indirect
-	sigs.k8s.io/controller-runtime v0.4.0
+	sigs.k8s.io/controller-runtime v0.5.0
 )
 
 // Pin the version of client-go to something that's compatible with katrogan's fork of api and apimachinery

flytepropeller/pkg/compiler/validators/interface.go

@@ -113,7 +113,7 @@ func ValidateUnderlyingInterface(w c.WorkflowBuilder, node c.NodeBuilder, errs e
 	case *core.Node_BranchNode:
 		iface, _ = validateBranchInterface(w, node, errs.NewScope())
 	default:
-		errs.Collect(errors.NewValueRequiredErr(node.GetId(), "Target"))
+		errs.Collect(errors.NewValueRequiredErr(node.GetId(), "TargetedPrefixString"))
 	}
 
 	if iface != nil {

flytepropeller/pkg/controller/nodes/task/handler.go

@@ -138,7 +138,7 @@ type Handler struct {
 	pluginRegistry  PluginRegistryIface
 	kubeClient      pluginCore.KubeClient
 	secretManager   pluginCore.SecretManager
-	resourceManager pluginCore.ResourceManager
+	resourceManager resourcemanager.BaseResourceManager
 	barrierCache    *barrier
 	cfg             *config.Config
 }

flytepropeller/pkg/controller/nodes/task/handler_test.go

@@ -303,7 +303,7 @@ func (t taskNodeStateHolder) PutDynamicNodeState(s handler.DynamicNodeState) err
 	panic("not implemented")
 }
 
-func CreateNoopResourceManager(ctx context.Context, scope promutils.Scope) pluginCore.ResourceManager {
+func CreateNoopResourceManager(ctx context.Context, scope promutils.Scope) resourcemanager.BaseResourceManager {
 	rmBuilder, _ := resourcemanager.GetResourceManagerBuilderByType(ctx, rmConfig.TypeNoop, scope)
 	rm, _ := rmBuilder.BuildResourceManager(ctx)
 	return rm

flytepropeller/pkg/controller/nodes/task/resourcemanager/config/config.go

@@ -29,7 +29,7 @@ var (
 type Config struct {
 	Type             Type        `json:"type" pflag:"noop,Which resource manager to use"`
 	ResourceMaxQuota int         `json:"resourceMaxQuota" pflag:",Global limit for concurrent Qubole queries"`
-	RedisConfig      RedisConfig `json:"redis" pflag:",Config for Redist resourcemanager."`
+	RedisConfig      RedisConfig `json:"redis" pflag:",Config for Redis resourcemanager."`
 }
 
 // Specific configs for Redis resource manager

flytepropeller/pkg/controller/nodes/task/resourcemanager/fullyqualifiedresourceconstraints.go

@@ -0,0 +1,45 @@
+package resourcemanager
+
+import (
+	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/core"
+	pluginCore "github.com/lyft/flyteplugins/go/tasks/pluginmachinery/core"
+)
+
+type ResourceConstraint interface {
+	IsAllowed(int64) bool
+}
+
+func isAllowed(constraintValue int64, actualValue int64) bool {
+	return constraintValue > actualValue
+}
+
+type BaseResourceConstraint struct {
+	Value int64
+}
+
+func (brc *BaseResourceConstraint) IsAllowed(actualValue int64) bool {
+	return isAllowed(brc.Value, actualValue)
+}
+
+type FullyQualifiedResourceConstraint struct {
+	TargetedPrefixString string
+	Value                int64
+}
+
+func (fqrc *FullyQualifiedResourceConstraint) IsAllowed(actualValue int64) bool {
+	return isAllowed(fqrc.Value, actualValue)
+}
+
+func composeFullyQualifiedProjectScopeResourceConstraint(spec pluginCore.ResourceConstraintsSpec, id *core.TaskExecutionIdentifier) FullyQualifiedResourceConstraint {
+	return FullyQualifiedResourceConstraint{
+		TargetedPrefixString: string(composeProjectScopePrefix(id)),
+		Value:                spec.ProjectScopeResourceConstraint.Value,
+	}
+}
+
+func composeFullyQualifiedNamespaceScopeResourceConstraint(spec pluginCore.ResourceConstraintsSpec, id *core.TaskExecutionIdentifier) FullyQualifiedResourceConstraint {
+	return FullyQualifiedResourceConstraint{
+		TargetedPrefixString: string(composeNamespaceScopePrefix(id)),
+		Value:                spec.NamespaceScopeResourceConstraint.Value,
+	}
+}