[GitHub] lburgazzoli closed pull request #56: Add a wait option to kamel run

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

[GitHub] lburgazzoli closed pull request #56: Add a wait option to kamel run

GitBox
lburgazzoli closed pull request #56: Add a wait option to kamel run
URL: https://github.com/apache/camel-k/pull/56
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/cmd/kamel/kamel.go b/cmd/kamel/kamel.go
index 3328370..ff56647 100644
--- a/cmd/kamel/kamel.go
+++ b/cmd/kamel/kamel.go
@@ -22,10 +22,12 @@ import (
  "github.com/apache/camel-k/pkg/client/cmd"
  _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
  "os"
+ "context"
 )
 
 func main() {
- rootCmd, err := cmd.NewKamelCommand()
+ ctx := context.Background()
+ rootCmd, err := cmd.NewKamelCommand(ctx)
  exitOnError(err)
 
  err = rootCmd.Execute()
diff --git a/deploy/operator.yaml b/deploy/operator.yaml
index a2c4abc..863b21c 100644
--- a/deploy/operator.yaml
+++ b/deploy/operator.yaml
@@ -15,6 +15,9 @@ spec:
       containers:
         - name: camel-k-operator
           image: docker.io/apache/camel-k:0.0.1-SNAPSHOT
+          ports:
+          - containerPort: 60000
+            name: metrics
           command:
           - camel-k-operator
           imagePullPolicy: Always
@@ -23,3 +26,5 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: OPERATOR_NAME
+              value: "camel-k-operator"
diff --git a/pkg/client/cmd/root.go b/pkg/client/cmd/root.go
index e0e2a08..50cfb40 100644
--- a/pkg/client/cmd/root.go
+++ b/pkg/client/cmd/root.go
@@ -23,15 +23,19 @@ import (
  "github.com/apache/camel-k/pkg/util/kubernetes"
  "github.com/pkg/errors"
  "github.com/spf13/cobra"
+ "context"
 )
 
 type RootCmdOptions struct {
+ Context    context.Context
  KubeConfig string
  Namespace  string
 }
 
-func NewKamelCommand() (*cobra.Command, error) {
- options := RootCmdOptions{}
+func NewKamelCommand(ctx context.Context) (*cobra.Command, error) {
+ options := RootCmdOptions{
+ Context: ctx,
+ }
  var cmd = cobra.Command{
  Use:   "kamel",
  Short: "Kamel is a awesome client tool for running Apache Camel integrations natively on Kubernetes",
diff --git a/pkg/client/cmd/run.go b/pkg/client/cmd/run.go
index d64a653..cee146e 100644
--- a/pkg/client/cmd/run.go
+++ b/pkg/client/cmd/run.go
@@ -31,6 +31,7 @@ import (
  "github.com/spf13/cobra"
  k8serrors "k8s.io/apimachinery/pkg/api/errors"
  "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "github.com/apache/camel-k/pkg/util/watch"
 )
 
 type RunCmdOptions struct {
@@ -38,6 +39,7 @@ type RunCmdOptions struct {
  Language        string
  IntegrationName string
  Dependencies    []string
+ Wait            bool
 }
 
 func NewCmdRun(rootCmdOptions *RootCmdOptions) *cobra.Command {
@@ -54,8 +56,10 @@ func NewCmdRun(rootCmdOptions *RootCmdOptions) *cobra.Command {
  }
 
  cmd.Flags().StringVarP(&options.Language, "language", "l", "", "Programming Language used to write the file")
- cmd.Flags().StringVarP(&options.IntegrationName, "name", "", "", "The integration name")
+ cmd.Flags().StringVar(&options.IntegrationName, "name", "", "The integration name")
  cmd.Flags().StringSliceVarP(&options.Dependencies, "dependency", "d", nil, "The integration dependency")
+ cmd.Flags().BoolVarP(&options.Wait, "wait", "w", false, "Waits for the integration to be running")
+ cmd.ParseFlags(os.Args)
 
  return &cmd
 }
@@ -74,10 +78,62 @@ func (*RunCmdOptions) validateArgs(cmd *cobra.Command, args []string) error {
 }
 
 func (o *RunCmdOptions) run(cmd *cobra.Command, args []string) error {
- code, err := o.loadCode(args[0])
+ integration, err := o.createIntegration(cmd, args)
  if err != nil {
  return err
  }
+ if o.Wait {
+ err = o.waitForIntegrationReady(integration)
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (o *RunCmdOptions) waitForIntegrationReady(integration *v1alpha1.Integration) error {
+ // Block this goroutine until the integration is in a final status
+ changes, err := watch.WatchStateChanges(o.Context, integration)
+ if err != nil {
+ return err
+ }
+
+ var lastStatusSeen *v1alpha1.IntegrationStatus
+
+watcher:
+ for {
+ select {
+ case <-o.Context.Done():
+ return nil
+ case i, ok := <-changes:
+ if !ok {
+ break watcher
+ }
+ lastStatusSeen = &i.Status
+ phase := string(i.Status.Phase)
+ if phase != "" {
+ fmt.Println("integration \""+integration.Name+"\" in phase", phase)
+ // TODO when we add health checks, we should wait until they are passed
+ if i.Status.Phase == v1alpha1.IntegrationPhaseRunning || i.Status.Phase == v1alpha1.IntegrationPhaseError {
+ // TODO display some error info when available in the status
+ break watcher
+ }
+ }
+ }
+ }
+
+ // TODO we may not be able to reach this state, since the build will be done without sources (until we add health checks)
+ if lastStatusSeen != nil && lastStatusSeen.Phase == v1alpha1.IntegrationPhaseError {
+ return errors.New("integration deployment failed")
+ }
+ return nil
+}
+
+func (o *RunCmdOptions) createIntegration(cmd *cobra.Command, args []string) (*v1alpha1.Integration, error) {
+ code, err := o.loadCode(args[0])
+ if err != nil {
+ return nil, err
+ }
 
  namespace := o.Namespace
 
@@ -124,14 +180,14 @@ func (o *RunCmdOptions) run(cmd *cobra.Command, args []string) error {
  clone := integration.DeepCopy()
  err = sdk.Get(clone)
  if err != nil {
- return err
+ return nil, err
  }
  integration.ResourceVersion = clone.ResourceVersion
  err = sdk.Update(&integration)
  }
 
  if err != nil {
- return err
+ return nil, err
  }
 
  if !existed {
@@ -139,7 +195,7 @@ func (o *RunCmdOptions) run(cmd *cobra.Command, args []string) error {
  } else {
  fmt.Printf("integration \"%s\" updated\n", name)
  }
- return nil
+ return &integration, nil
 }
 
 func (*RunCmdOptions) loadCode(fileName string) (string, error) {
diff --git a/pkg/util/watch/watch.go b/pkg/util/watch/watch.go
new file mode 100644
index 0000000..3ec0bda
--- /dev/null
+++ b/pkg/util/watch/watch.go
@@ -0,0 +1,82 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package watch
+
+import (
+ "github.com/apache/camel-k/pkg/apis/camel/v1alpha1"
+ "context"
+ "github.com/operator-framework/operator-sdk/pkg/k8sclient"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+ "github.com/operator-framework/operator-sdk/pkg/util/k8sutil"
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+ "github.com/sirupsen/logrus"
+)
+
+// Watches a integration resource and send it through a channel when its status changes
+func WatchStateChanges(ctx context.Context, integration *v1alpha1.Integration) (<-chan *v1alpha1.Integration, error) {
+ resourceClient, _, err := k8sclient.GetResourceClient(integration.APIVersion, integration.Kind, integration.Namespace)
+ if err != nil {
+ return nil, err
+ }
+ watcher, err := resourceClient.Watch(metav1.ListOptions{
+ FieldSelector: "metadata.name=" + integration.Name,
+ })
+ if err != nil {
+ return nil, err
+ }
+ events := watcher.ResultChan()
+
+ out := make(chan *v1alpha1.Integration)
+ var lastObservedState *v1alpha1.IntegrationPhase
+
+ go func() {
+ defer close(out)
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case e, ok := <-events:
+ if !ok {
+ return
+ }
+
+ if e.Object != nil {
+ if runtimeUnstructured, ok := e.Object.(runtime.Unstructured); ok {
+ unstr := unstructured.Unstructured{
+ Object: runtimeUnstructured.UnstructuredContent(),
+ }
+ icopy := integration.DeepCopy()
+ err := k8sutil.UnstructuredIntoRuntimeObject(&unstr, icopy)
+ if err != nil {
+ logrus.Error("Unexpected error detected when watching resource", err)
+ return // closes the channel
+ }
+
+ if lastObservedState == nil || *lastObservedState != icopy.Status.Phase {
+ lastObservedState = &icopy.Status.Phase
+ out <- icopy
+ }
+ }
+ }
+ }
+ }
+ }()
+
+ return out, nil
+}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
[hidden email]


With regards,
Apache Git Services