Merge pull request #10 from wpjunior/trigger-job

Trigger acl-operator for tsuru jobs

engine/operator/engine.go

@@ -13,7 +13,6 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/tsuru/acl-api/api/types"
 	"github.com/tsuru/acl-api/engine"
-	"github.com/tsuru/acl-api/kubernetes"
 	aclKube "github.com/tsuru/acl-api/kubernetes"
 	"github.com/tsuru/acl-api/rule"
 	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
@@ -52,13 +51,21 @@ func (e *ACLOperatorEngine) AfterSync() error {
 }
 
 func (e *ACLOperatorEngine) Sync(r types.Rule) (interface{}, error) {
-	ctx := context.TODO()
-	log := logger.WithField("ruleid", r.RuleID)
+	if r.Source.TsuruApp != nil {
+		return e.SyncApp(r)
+	}
 
-	if r.Source.TsuruApp == nil {
-		return nil, nil
+	if r.Source.TsuruJob != nil {
+		return e.SyncJob(r)
 	}
 
+	return nil, nil
+}
+
+func (e *ACLOperatorEngine) SyncApp(r types.Rule) (interface{}, error) {
+	ctx := context.TODO()
+	log := logger.WithField("ruleid", r.RuleID)
+
 	source, err := e.logicCache.LogicFromRule(r)
 	if err != nil {
 		return nil, err
@@ -67,7 +74,7 @@ func (e *ACLOperatorEngine) Sync(r types.Rule) (interface{}, error) {
 		return nil, nil
 	}
 
-	restConfig, err := source.KubernetesRestConfig()
+	restConfig, _, err := source.KubernetesRestConfig()
 	if err != nil {
 		return nil, err
 	}
@@ -77,7 +84,7 @@ func (e *ACLOperatorEngine) Sync(r types.Rule) (interface{}, error) {
 		return nil, nil
 	}
 
-	tsuruClient, err := kubernetes.GetTsuruClientWithRestConfig(restConfig)
+	tsuruClient, err := aclKube.GetTsuruClientWithRestConfig(restConfig)
 	if err != nil {
 		return "", err
 	}
@@ -145,3 +152,94 @@ func (e *ACLOperatorEngine) Sync(r types.Rule) (interface{}, error) {
 
 	return "triggered acl-operator in the last minute", nil
 }
+
+func (e *ACLOperatorEngine) SyncJob(r types.Rule) (interface{}, error) {
+	ctx := context.TODO()
+	log := logger.WithField("ruleid", r.RuleID)
+
+	source, err := e.logicCache.LogicFromRule(r)
+	if err != nil {
+		return nil, err
+	}
+
+	if source == nil {
+		return nil, nil
+	}
+
+	restConfig, pool, err := source.KubernetesRestConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	if restConfig == nil {
+		log.Debugf("Ignoring rule, not a kubernetes source")
+		return nil, nil
+	}
+
+	k8sClient, err := aclKube.GetClientWithRestConfig(restConfig)
+	if err != nil {
+		return "", err
+	}
+
+	tsuruJobName := r.Source.TsuruJob.JobName
+	cronJobNamespace := k8sClient.BatchV1().CronJobs("tsuru-" + pool)
+	cronJobCRD, err := cronJobNamespace.Get(ctx, tsuruJobName, metav1.GetOptions{})
+	if err != nil {
+		if k8sErrors.IsNotFound(err) {
+			return "", nil
+		}
+		return "", err
+	}
+
+	originalData, err := json.Marshal(cronJobCRD)
+	if err != nil {
+		return "", err
+	}
+
+	lastUpdatedStr := cronJobCRD.Annotations[lastUpdatedAnnotation]
+	var lastUpdated time.Time
+	needsUpdate := false
+
+	if lastUpdatedStr == "" {
+		needsUpdate = true
+	} else {
+		lastUpdated, err = time.Parse(time.RFC3339, lastUpdatedStr)
+		if err != nil {
+			return "", err
+		}
+
+		if r.Created.UTC().Add(time.Minute).After(lastUpdated) {
+			needsUpdate = true
+		}
+
+		if time.Now().UTC().After(lastUpdated.Add(time.Minute)) {
+			needsUpdate = true
+		}
+	}
+
+	if needsUpdate {
+		if cronJobCRD.Annotations == nil {
+			cronJobCRD.Annotations = map[string]string{}
+		}
+		cronJobCRD.Annotations[lastUpdatedAnnotation] = time.Now().UTC().Format(time.RFC3339)
+
+		updatedData, err := json.Marshal(cronJobCRD)
+		if err != nil {
+			return "", err
+		}
+
+		data, err := jsonpatch.CreateMergePatch(originalData, updatedData)
+		if err != nil {
+			return "", err
+		}
+
+		_, err = cronJobNamespace.Patch(ctx, tsuruJobName, k8sTypes.MergePatchType, data, metav1.PatchOptions{})
+		if err != nil {
+			return "", err
+		}
+
+		return "triggered acl-operator", nil
+	}
+
+	return "triggered acl-operator in the last minute", nil
+}