3619

pending completion

Build # 3619

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

Merge pull request #1075 from stanford-oval/wip/fixes

Misc bug fixes

Run Details

2145 of 3909 branches covered (54.87%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 2 files covered. (100.0%)

1 existing line in 1 file now uncovered.

8246 of 11916 relevant lines covered (69.2%)

583.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.08

/go/k8s/controllers/user_controller.go

// Copyright 2021 The Board of Trustees of the Leland Stanford Junior University
//
// Licensed 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 controllers

import (
        "bytes"
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "net/http"
        "path"
        "strconv"
        "strings"
        "time"

        logging "log"

        "github.com/go-logr/logr"
        "gorm.io/gorm"
        appsv1 "k8s.io/api/apps/v1"
        corev1 "k8s.io/api/core/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

        apierrors "k8s.io/apimachinery/pkg/api/errors"
        "k8s.io/apimachinery/pkg/runtime"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/client"

        "almond-cloud/config"
        "almond-cloud/dbproxy"
        backendv1 "almond-cloud/k8s/api/v1"
        "almond-cloud/sql"
)

// UserReconciler reconciles a User object
type UserReconciler struct {
        client.Client
        Scheme                      *runtime.Scheme
        Log                         logr.Logger
        almondConfig                *config.AlmondConfig
        localCache                  map[string]CacheEntry
        developerDeploymentTemplate appsv1.Deployment
        developerServiceTemplate    corev1.Service
}

// UserState constants
type UserState string

const (
        Starting UserState = "starting"
        Running  UserState = "running"
        Idle     UserState = "idle"
        Stopping UserState = "stopping"
)

// PlatformOptions is part of runEngine request
type PlatformOptions struct {
        UserID             int64   `json:"userId"`
        CloudID            string  `json:"cloudId"`
        AuthToken          string  `json:"authToken"`
        DeveloperKey       *string `json:"developerKey"`
        Locale             string  `json:"locale"`
        Timezone           string  `json:"timezone"`
        DBProxyURL         string  `json:"dbProxyUrl"`
        DBProxyAccessToken string  `json:"dbProxyAccessToken"`
        HumanName          *string `json:"humanName"`
        Email              *string `json:"email"`
}

func platformOptions(user *sql.User, developerKey *string, dbProxyURL string, dbProxyToken string) *PlatformOptions {
        return &PlatformOptions{
                UserID:             user.ID,
                CloudID:            user.CloudID,
                AuthToken:          user.AuthToken,
                DeveloperKey:       developerKey,
                Locale:             user.Locale,
                Timezone:           user.Timezone,
                DBProxyURL:         dbProxyURL,
                DBProxyAccessToken: dbProxyToken,
                HumanName:          user.HumanName,
                Email:              user.Email,
        }
}

// JSONResposne from http service
type JSONResponse struct {
        Result string      `json:"result"`
        Data   interface{} `json:"data"`
}

type CacheEntry struct {
        Value      interface{}
        Expiration time.Time
}

const kDefaultCacheDuration = 2 * time.Hour

type dbQueryFunc func(db *gorm.DB, userID int64) (interface{}, error)

const kTrustedDeveloper int = 8

// NewUserReconciler initializes UserReconciler and reads template files from configDir.
func NewUserReconciler(client client.Client, scheme *runtime.Scheme, log logr.Logger, almondConfig *config.AlmondConfig, configDir string) *UserReconciler {
        r := &UserReconciler{
                Client:       client,
                Scheme:       scheme,
                Log:          log,
                almondConfig: almondConfig,
                localCache:   make(map[string]CacheEntry),
        }
        if err := ReadJSONFile(path.Join(configDir, "developer-deployment.json"), &r.developerDeploymentTemplate); err != nil {
                logging.Fatal(err)
        }
        if err := ReadJSONFile(path.Join(configDir, "developer-service.json"), &r.developerServiceTemplate); err != nil {
                logging.Fatal(err)
        }
        return r
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.8.3/pkg/reconcile
func (r *UserReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        _ = r.Log.WithValues("user-controller", req.NamespacedName)
        r.Log.Info("--- start ---")
        ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
        defer cancel()

        var (
                err              error
                userID           int64
                trustedDeveloper bool
                user             *backendv1.User
                stop             bool
                result           ctrl.Result
                currentStatus    backendv1.UserStatus
        )

        defer func() {
                if user != nil {
                        if len(user.Spec.Mode) == 0 {
                                if r.almondConfig.EnableDeveloperBackend && trustedDeveloper {
                                        user.Spec.Mode = "developer"
                                } else {
                                        user.Spec.Mode = "shared"
                                }
                                r.Update(ctx, user)
                        }
                        user.Status = currentStatus
                        r.Status().Update(ctx, user)
                        r.Log.Info("update status:", "user", user.Spec.ID, "status", user.Status)
                }
                r.Log.Info("--- end ---")
        }()

        userID, trustedDeveloper, err = r.getUserFromName(req.Name)
        if err != nil {
                return ctrl.Result{}, err
        }

        if r.almondConfig.EnableDeveloperBackend && trustedDeveloper {
                user, currentStatus, stop, result, err = r.handleDeveloper(ctx, req, userID)
        } else {
                user, currentStatus, stop, result, err = r.handleSharedUser(ctx, req, userID)
        }
        if err != nil || stop {
                return result, err
        }

        if len(user.Status.Backend) > 0 && user.Status.Backend != currentStatus.Backend {
                // backend url has changed, kill the old backend
                r.Log.Info("backends changed:", "user", user.Spec.ID, "old", user.Status.Backend, "new", currentStatus.Backend)
                if err = r.killEngine(ctx, userID, user.Status.Backend); err != nil {
                        r.Log.Error(err, "kill old engine failed")
                }
        }

        if UserState(currentStatus.State) == Running {
                return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
        }

        if UserState(currentStatus.State) == Idle {
                r.Log.Info("delete idle engine:", "user", user.Spec.ID)
                if err = r.killEngine(ctx, userID, currentStatus.Backend); err != nil {
                        r.Log.Error(err, "kill idle engine failed")
                }
                return ctrl.Result{}, r.Client.Delete(ctx, user)
        }

        if err = r.runEngine(ctx, user.Spec.ID, currentStatus.Backend); err != nil {
                currentStatus.State = err.Error()
                return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
        }
        currentStatus.State = string(Starting)
        return ctrl.Result{RequeueAfter: 1 * time.Second}, nil
}

func (r *UserReconciler) getUserFromName(name string) (uid int64, trustedDeveloper bool, err error) {
        uid, err = strconv.ParseInt(strings.TrimPrefix(name, "user-"), 10, 64)
        if err != nil {
                return
        }
        v, err := r.getDBEntry("user", uid, getUser, true)
        if err != nil {
                return
        }
        user := v.(*sql.User)
        if user.Roles&kTrustedDeveloper != 0 {
                trustedDeveloper = true
        }
        return
}

func (r *UserReconciler) handleDeveloper(ctx context.Context, req ctrl.Request,
        userID int64) (user *backendv1.User, currentStatus backendv1.UserStatus, stop bool, result ctrl.Result, err error) {
        user = &backendv1.User{}
        // for developer users, make sure deployment and service are up before proceeding.
        if err = r.Client.Get(ctx, req.NamespacedName, user); err != nil {
                if apierrors.IsNotFound(err) {
                        if err = r.deleteDeploymentService(ctx, req, userID); err != nil {
                                r.Log.Error(err, "fail to delete developer deployment or service")
                        }
                }
                stop = true
                user = nil
                return
        }
        if !user.ObjectMeta.DeletionTimestamp.IsZero() {
                // user is marked for deletion
                currentStatus.State = string(Stopping)
                if err = r.deleteDeploymentService(ctx, req, userID); err != nil {
                        r.Log.Error(err, "fail to delete developer deployment or service marked for deletion")
                }
                stop = true
                return
        }
        deployment := &appsv1.Deployment{}
        if err = r.Client.Get(ctx, req.NamespacedName, deployment); err != nil {
                if apierrors.IsNotFound(err) {
                        if err = r.createDeployment(ctx, req.Name, req.Namespace); err != nil {
                                return
                        }
                        stop = true
                        result = ctrl.Result{RequeueAfter: 2 * time.Second}
                        // setting err to nil so ControllerManager will respect the RequeueAfter time
                        err = nil
                        return
                }
                return
        }
        if deployment.Status.AvailableReplicas <= 0 {
                currentStatus.State = string(Starting)
                stop = true
                result = ctrl.Result{RequeueAfter: 2 * time.Second}
                return
        }
        service := &corev1.Service{}
        if err = r.Client.Get(ctx, req.NamespacedName, service); err != nil {
                if apierrors.IsNotFound(err) {
                        if err = r.createService(ctx, req.Name, req.Namespace); err != nil {
                                return
                        }
                        currentStatus.State = string(Starting)
                        stop = true
                        result = ctrl.Result{RequeueAfter: 2 * time.Second}
                        err = nil
                        return
                }
                return
        }
        if len(service.Spec.ClusterIP) == 0 {
                currentStatus.State = string(Starting)
                stop = true
                result = ctrl.Result{RequeueAfter: 2 * time.Second}
                err = nil
                return
        }
        if len(service.Spec.Ports) == 0 {
                err = errors.New("service port not configured")
                return
        }
        currentStatus.Backend = fmt.Sprintf("http://%s:%d", service.Spec.ClusterIP, service.Spec.Ports[0].Port)

        engineStatus, err := r.engineStatus(ctx, userID, currentStatus.Backend)
        if err != nil {
                currentStatus.State = err.Error()
                if isDialError(err) {
                        stop = true
                        err = nil
                        result = ctrl.Result{RequeueAfter: 2 * time.Second}
                        return
                }
                return
        }
        currentStatus.State = string(engineStatus)
        return
}

func (r *UserReconciler) handleSharedUser(ctx context.Context, req ctrl.Request,
        userID int64) (user *backendv1.User, currentStatus backendv1.UserStatus, stop bool, result ctrl.Result, err error) {
        currentStatus.Backend, err = r.getSharedBackendURL(ctx, req.Namespace, userID)
        if err != nil {
                currentStatus.State = err.Error()
                result = ctrl.Result{RequeueAfter: 2 * time.Second}
                stop = true
                err = nil
                return
        }

        engineStatus, err := r.engineStatus(ctx, userID, currentStatus.Backend)
        if err != nil {
                currentStatus.State = err.Error()
                if isDialError(err) {
                        stop = true
                        err = nil
                        result = ctrl.Result{RequeueAfter: 2 * time.Second}
                        return
                }
                return
        }
        currentStatus.State = string(engineStatus)

        user = &backendv1.User{}
        if err = r.Client.Get(ctx, req.NamespacedName, user); err != nil {
                if apierrors.IsNotFound(err) {
                        // User is already deleted, kill engine if it's still running.
                        if engineStatus == Running || engineStatus == Idle {
                                r.Log.Info("kill engine for already deleted user:", "user", userID)
                                err = r.killEngine(ctx, userID, currentStatus.Backend)
                                return
                        }
                        err = nil
                        stop = true
                }
                return
        }

        if !user.ObjectMeta.DeletionTimestamp.IsZero() {
                // object is marked for deletion
                if engineStatus == Running || engineStatus == Idle {
                        r.Log.Info("kill engine for user marked for deletion:", "user", userID)
                        if err = r.killEngine(ctx, userID, currentStatus.Backend); err != nil {
                                r.Log.Error(err, "fail to kill engine marked for deletion")
                        }
                }
                stop = true
                return
        }
        return
}

func (r *UserReconciler) getSharedBackendURL(ctx context.Context, namespace string, uid int64) (urlStr string, err error) {
        endpoints := &corev1.Endpoints{}
        // fetch backend endpoints.
        if err = r.Get(ctx, client.ObjectKey{Namespace: namespace, Name: "shared-backend"}, endpoints); err != nil {
                return
        }
        // StatefulSet endpoints are always ordered by name.
        var backendURLs []string
        for _, subset := range endpoints.Subsets {
                for _, addr := range subset.Addresses {
                        u := fmt.Sprintf("http://%s:%d", addr.IP, subset.Ports[0].Port)
                        backendURLs = append(backendURLs, u)
                }
        }
        if len(backendURLs) == 0 {
                err = fmt.Errorf("backend endpoints not found")
        } else {
                urlStr = backendURLs[uid%int64(len(backendURLs))]
        }
        return
}

func (r *UserReconciler) createDeployment(ctx context.Context, name, namespace string) error {
        deployment := NewDeployment(&r.developerDeploymentTemplate, name, namespace)
        if err := r.Client.Create(ctx, deployment); err != nil {
                return err
        }
        return nil
}

func (r *UserReconciler) createService(ctx context.Context, name, namespace string) error {
        service := NewService(&r.developerServiceTemplate, name, namespace)
        if err := r.Client.Create(ctx, service); err != nil {
                return err
        }
        return nil
}

func (r *UserReconciler) getDBEntry(keyPrefix string, userID int64, fn dbQueryFunc, useCache bool) (interface{}, error) {
        cacheKey := fmt.Sprintf("%s-%d", keyPrefix, userID)
        cacheEntry, ok := r.localCache[cacheKey]
        if useCache && ok && cacheEntry.Expiration.After(time.Now()) {
                return cacheEntry.Value, nil
        }
        db := sql.GetDB()
        v, err := fn(db, userID)
        if err != nil {
                return nil, err
        }
        r.localCache[cacheKey] = CacheEntry{v, time.Now().Add(kDefaultCacheDuration)}
        return v, nil
}

func getUser(db *gorm.DB, userID int64) (interface{}, error) {
        return sql.GetUser(db, userID)
}

func getDeveloperKey(db *gorm.DB, userID int64) (interface{}, error) {
        return sql.GetDeveloperKey(db, userID)
}

func (r *UserReconciler) httpWithContext(ctx context.Context, method, url string, body io.Reader) (*http.Response, error) {
        r.Log.Info(fmt.Sprintf("HTTP %v url:%v", method, url))
        req, err := http.NewRequestWithContext(ctx, method, url, body)
        if err != nil {
                return nil, err
        }
        if body != nil {
                req.Header.Set("Content-Type", "application/json")
        }
        resp, err := http.DefaultClient.Do(req)
        return resp, err
}

func (r *UserReconciler) engineStatus(ctx context.Context, userID int64, userURL string) (UserState, error) {
        resp, err := r.httpWithContext(ctx, "GET", fmt.Sprintf("%s/engine-status?userid=%d", userURL, userID), nil)
        if err != nil {
                return "", err
        }
        defer resp.Body.Close()
        if resp.StatusCode != http.StatusOK {
                return "", fmt.Errorf("Failed to get engine status: %+v", resp)
        }
        jsonResponse := &JSONResponse{Data: ""}
        if err := json.NewDecoder(resp.Body).Decode(jsonResponse); err != nil {
                return "", err
        }
        r.Log.Info("engine status:", "user", userID, "resp", jsonResponse)
        return UserState(jsonResponse.Data.(string)), nil
}

func (r *UserReconciler) killEngine(ctx context.Context, userID int64, backendURL string) error {
        resp, err := r.httpWithContext(ctx, "GET", fmt.Sprintf("%s/kill-engine?userid=%d", backendURL, userID), nil)
        if err != nil {
                return err
        }
        if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("Failed killEngine http status: %+v", resp)
        }
        return nil
}

func (r *UserReconciler) runEngine(ctx context.Context, userID int64, userURL string) error {
        v, err := r.getDBEntry("user", userID, getUser, false)
        if err != nil {
                return err
        }
        u := v.(*sql.User)
        v, err = r.getDBEntry("developer-key", userID, getDeveloperKey, false)
        if err != nil {
                return err
        }
        developerKey := v.(*string)

        token, err := dbproxy.SignToken(u.ID)
        if err != nil {
                return err
        }

        options := platformOptions(u, developerKey, config.GetAlmondConfig().DatabaseProxyURL, token)

        b, err := json.Marshal(options)
        if err != nil {
                return err
        }
        resp, err := r.httpWithContext(ctx, "POST", fmt.Sprintf("%s/run-engine", userURL), bytes.NewReader(b))
        if err != nil {
                return err
        }
        defer resp.Body.Close()
        if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("Failed runEngine http status: %+v", resp)
        }

        return nil
}

func (r *UserReconciler) deleteDeploymentService(ctx context.Context, req ctrl.Request, userID int64) error {
        r.Log.Info("Deleting deployment and service:", "user", userID)
        meta := &metav1.ObjectMeta{Name: req.Name, Namespace: req.Namespace}
        err1 := r.deleteService(ctx, req, userID, meta)
        err2 := r.deleteDeployment(ctx, req, meta)
        if err1 != nil && err2 != nil {
                return fmt.Errorf("Error1:%v\nError2:%v", err1, err2)
        }
        if err1 != nil {
                return err1
        }
        if err2 != nil {
                return err2
        }
        return nil
}

func (r *UserReconciler) deleteDeployment(ctx context.Context, req ctrl.Request, meta *metav1.ObjectMeta) error {
        deployment := &appsv1.Deployment{ObjectMeta: *meta}
        if err := r.Client.Get(ctx, req.NamespacedName, deployment); err != nil {
                if apierrors.IsNotFound(err) {
                        return nil
                }
        }
        if err := r.Client.Delete(ctx, deployment); err != nil {
                return err
        }
        return nil
}

func (r *UserReconciler) deleteService(ctx context.Context, req ctrl.Request, userID int64, meta *metav1.ObjectMeta) error {
        service := &corev1.Service{ObjectMeta: *meta}
        if err := r.Client.Get(ctx, req.NamespacedName, service); err != nil {
                if apierrors.IsNotFound(err) {
                        return nil
                }
        }
        if len(service.Spec.ClusterIP) > 0 && len(service.Spec.Ports) > 0 {
                backend := fmt.Sprintf("http://%s:%d", service.Spec.ClusterIP, service.Spec.Ports[0].Port)
                if err := r.killEngine(ctx, userID, backend); err != nil {
                        r.Log.Error(err, "kill engine failed")
                }
        }
        if err := r.Client.Delete(ctx, service); err != nil {
                return err
        }
        return nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *UserReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&backendv1.User{}).
                Complete(r)
}

1	// Copyright 2021 The Board of Trustees of the Leland Stanford Junior University
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package controllers
16
17	import (
18	"bytes"
19	"context"
20	"encoding/json"
21	"errors"
22	"fmt"
23	"io"
24	"net/http"
25	"path"
26	"strconv"
27	"strings"
28	"time"
29
30	logging "log"
31
32	"github.com/go-logr/logr"
33	"gorm.io/gorm"
34	appsv1 "k8s.io/api/apps/v1"
35	corev1 "k8s.io/api/core/v1"
36	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
37
38	apierrors "k8s.io/apimachinery/pkg/api/errors"
39	"k8s.io/apimachinery/pkg/runtime"
40	ctrl "sigs.k8s.io/controller-runtime"
41	"sigs.k8s.io/controller-runtime/pkg/client"
42
43	"almond-cloud/config"
44	"almond-cloud/dbproxy"
45	backendv1 "almond-cloud/k8s/api/v1"
46	"almond-cloud/sql"
47	)
48
49	// UserReconciler reconciles a User object
50	type UserReconciler struct {
51	client.Client
52	Scheme *runtime.Scheme
53	Log logr.Logger
54	almondConfig *config.AlmondConfig
55	localCache map[string]CacheEntry
56	developerDeploymentTemplate appsv1.Deployment
57	developerServiceTemplate corev1.Service
58	}
59
60	// UserState constants
61	type UserState string
62
63	const (
64	Starting UserState = "starting"
65	Running UserState = "running"
66	Idle UserState = "idle"
67	Stopping UserState = "stopping"
68	)
69
70	// PlatformOptions is part of runEngine request
71	type PlatformOptions struct {
72	UserID int64 `json:"userId"`
73	CloudID string `json:"cloudId"`
74	AuthToken string `json:"authToken"`
75	DeveloperKey *string `json:"developerKey"`
76	Locale string `json:"locale"`
77	Timezone string `json:"timezone"`
78	DBProxyURL string `json:"dbProxyUrl"`
79	DBProxyAccessToken string `json:"dbProxyAccessToken"`
80	HumanName *string `json:"humanName"`
81	Email *string `json:"email"`
82	}
83
84	func platformOptions(user sql.User, developerKey string, dbProxyURL string, dbProxyToken string) *PlatformOptions {	40✔
85	return &PlatformOptions{	40✔
86	UserID: user.ID,	40✔
87	CloudID: user.CloudID,	40✔
88	AuthToken: user.AuthToken,	40✔
89	DeveloperKey: developerKey,	40✔
90	Locale: user.Locale,	40✔
91	Timezone: user.Timezone,	40✔
92	DBProxyURL: dbProxyURL,	40✔
93	DBProxyAccessToken: dbProxyToken,	40✔
94	HumanName: user.HumanName,	40✔
95	Email: user.Email,	40✔
96	}	40✔
97	}	40✔
98
99	// JSONResposne from http service
100	type JSONResponse struct {
101	Result string `json:"result"`
102	Data interface{} `json:"data"`
103	}
104
105	type CacheEntry struct {
106	Value interface{}
107	Expiration time.Time
108	}
109
110	const kDefaultCacheDuration = 2 * time.Hour
111
112	type dbQueryFunc func(db *gorm.DB, userID int64) (interface{}, error)
113
114	const kTrustedDeveloper int = 8
115
116	// NewUserReconciler initializes UserReconciler and reads template files from configDir.
117	func NewUserReconciler(client client.Client, scheme runtime.Scheme, log logr.Logger, almondConfig config.AlmondConfig, configDir string) *UserReconciler {	1✔
118	r := &UserReconciler{	1✔
119	Client: client,	1✔
120	Scheme: scheme,	1✔
121	Log: log,	1✔
122	almondConfig: almondConfig,	1✔
123	localCache: make(map[string]CacheEntry),	1✔
124	}	1✔
125	if err := ReadJSONFile(path.Join(configDir, "developer-deployment.json"), &r.developerDeploymentTemplate); err != nil {	1✔
126	logging.Fatal(err)	×
127	}	×
128	if err := ReadJSONFile(path.Join(configDir, "developer-service.json"), &r.developerServiceTemplate); err != nil {	1✔
129	logging.Fatal(err)	×
130	}	×
131	return r	1✔
132	}
133
134	// Reconcile is part of the main kubernetes reconciliation loop which aims to
135	// move the current state of the cluster closer to the desired state.
136	// For more details, check Reconcile and its Result here:
137	// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.8.3/pkg/reconcile
138	func (r *UserReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	255✔
139	_ = r.Log.WithValues("user-controller", req.NamespacedName)	255✔
140	r.Log.Info("--- start ---")	255✔
141	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)	255✔
142	defer cancel()	255✔
143		255✔
144	var (	255✔
145	err error	255✔
146	userID int64	255✔
147	trustedDeveloper bool	255✔
148	user *backendv1.User	255✔
149	stop bool	255✔
150	result ctrl.Result	255✔
151	currentStatus backendv1.UserStatus	255✔
152	)	255✔
153		255✔
154	defer func() {	510✔
155	if user != nil {	497✔
156	if len(user.Spec.Mode) == 0 {	279✔
157	if r.almondConfig.EnableDeveloperBackend && trustedDeveloper {	42✔
158	user.Spec.Mode = "developer"	5✔
159	} else {	37✔
160	user.Spec.Mode = "shared"	32✔
161	}	32✔
162	r.Update(ctx, user)	37✔
163	}
164	user.Status = currentStatus	242✔
165	r.Status().Update(ctx, user)	242✔
166	r.Log.Info("update status:", "user", user.Spec.ID, "status", user.Status)	242✔
167	}
168	r.Log.Info("--- end ---")	255✔
169	}()
170
171	userID, trustedDeveloper, err = r.getUserFromName(req.Name)	255✔
172	if err != nil {	255✔
173	return ctrl.Result{}, err	×
174	}	×
175
176	if r.almondConfig.EnableDeveloperBackend && trustedDeveloper {	413✔
177	user, currentStatus, stop, result, err = r.handleDeveloper(ctx, req, userID)	158✔
178	} else {	255✔
179	user, currentStatus, stop, result, err = r.handleSharedUser(ctx, req, userID)	97✔
180	}	97✔
181	if err != nil \|\| stop {	344✔
182	return result, err	89✔
183	}	89✔
184
185	if len(user.Status.Backend) > 0 && user.Status.Backend != currentStatus.Backend {	166✔
186	// backend url has changed, kill the old backend	×
187	r.Log.Info("backends changed:", "user", user.Spec.ID, "old", user.Status.Backend, "new", currentStatus.Backend)	×
188	if err = r.killEngine(ctx, userID, user.Status.Backend); err != nil {	×
189	r.Log.Error(err, "kill old engine failed")	×
190	}	×
191	}
192
193	if UserState(currentStatus.State) == Running {	285✔
194	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil	119✔
195	}	119✔
196
197	if UserState(currentStatus.State) == Idle {	54✔
198	r.Log.Info("delete idle engine:", "user", user.Spec.ID)	7✔
199	if err = r.killEngine(ctx, userID, currentStatus.Backend); err != nil {	7✔
200	r.Log.Error(err, "kill idle engine failed")	×
201	}	×
202	return ctrl.Result{}, r.Client.Delete(ctx, user)	7✔
203	}
204
205	if err = r.runEngine(ctx, user.Spec.ID, currentStatus.Backend); err != nil {	40✔
206	currentStatus.State = err.Error()	×
207	return ctrl.Result{RequeueAfter: 2 * time.Second}, nil	×
208	}	×
209	currentStatus.State = string(Starting)	40✔
210	return ctrl.Result{RequeueAfter: 1 * time.Second}, nil	40✔
211	}
212
213	func (r *UserReconciler) getUserFromName(name string) (uid int64, trustedDeveloper bool, err error) {	255✔
214	uid, err = strconv.ParseInt(strings.TrimPrefix(name, "user-"), 10, 64)	255✔
215	if err != nil {	255✔
216	return	×
217	}	×
218	v, err := r.getDBEntry("user", uid, getUser, true)	255✔
219	if err != nil {	255✔
220	return	×
221	}	×
222	user := v.(*sql.User)	255✔
223	if user.Roles&kTrustedDeveloper != 0 {	413✔
224	trustedDeveloper = true	158✔
225	}	158✔
226	return	255✔
227	}
228
229	func (r *UserReconciler) handleDeveloper(ctx context.Context, req ctrl.Request,
230	userID int64) (user *backendv1.User, currentStatus backendv1.UserStatus, stop bool, result ctrl.Result, err error) {	158✔
231	user = &backendv1.User{}	158✔
232	// for developer users, make sure deployment and service are up before proceeding.	158✔
233	if err = r.Client.Get(ctx, req.NamespacedName, user); err != nil {	163✔
234	if apierrors.IsNotFound(err) {	10✔
235	if err = r.deleteDeploymentService(ctx, req, userID); err != nil {	5✔
236	r.Log.Error(err, "fail to delete developer deployment or service")	×
237	}	×
238	}
239	stop = true	5✔
240	user = nil	5✔
241	return	5✔
242	}
243	if !user.ObjectMeta.DeletionTimestamp.IsZero() {	153✔
244	// user is marked for deletion	×
245	currentStatus.State = string(Stopping)	×
246	if err = r.deleteDeploymentService(ctx, req, userID); err != nil {	×
247	r.Log.Error(err, "fail to delete developer deployment or service marked for deletion")	×
248	}	×
249	stop = true	×
250	return	×
251	}
252	deployment := &appsv1.Deployment{}	153✔
253	if err = r.Client.Get(ctx, req.NamespacedName, deployment); err != nil {	158✔
254	if apierrors.IsNotFound(err) {	10✔
255	if err = r.createDeployment(ctx, req.Name, req.Namespace); err != nil {	5✔
256	return	×
257	}	×
258	stop = true	5✔
259	result = ctrl.Result{RequeueAfter: 2 * time.Second}	5✔
260	// setting err to nil so ControllerManager will respect the RequeueAfter time	5✔
261	err = nil	5✔
262	return	5✔
263	}
264	return	×
265	}
266	if deployment.Status.AvailableReplicas <= 0 {	163✔
267	currentStatus.State = string(Starting)	15✔
268	stop = true	15✔
269	result = ctrl.Result{RequeueAfter: 2 * time.Second}	15✔
270	return	15✔
271	}	15✔
272	service := &corev1.Service{}	133✔
273	if err = r.Client.Get(ctx, req.NamespacedName, service); err != nil {	138✔
274	if apierrors.IsNotFound(err) {	10✔
275	if err = r.createService(ctx, req.Name, req.Namespace); err != nil {	5✔
276	return	×
277	}	×
278	currentStatus.State = string(Starting)	5✔
279	stop = true	5✔
280	result = ctrl.Result{RequeueAfter: 2 * time.Second}	5✔
281	err = nil	5✔
282	return	5✔
283	}
284	return	×
285	}
286	if len(service.Spec.ClusterIP) == 0 {	128✔
287	currentStatus.State = string(Starting)	×
288	stop = true	×
289	result = ctrl.Result{RequeueAfter: 2 * time.Second}	×
290	err = nil	×
291	return	×
292	}	×
293	if len(service.Spec.Ports) == 0 {	128✔
294	err = errors.New("service port not configured")	×
295	return	×
296	}	×
297	currentStatus.Backend = fmt.Sprintf("http://%s:%d", service.Spec.ClusterIP, service.Spec.Ports[0].Port)	128✔
298		128✔
299	engineStatus, err := r.engineStatus(ctx, userID, currentStatus.Backend)	128✔
300	if err != nil {	167✔
301	currentStatus.State = err.Error()	39✔
302	if isDialError(err) {	78✔
303	stop = true	39✔
304	err = nil	39✔
305	result = ctrl.Result{RequeueAfter: 2 * time.Second}	39✔
306	return	39✔
307	}	39✔
UNCOV 308	return	×
309	}
310	currentStatus.State = string(engineStatus)	89✔
311	return	89✔
312	}
313
314	func (r *UserReconciler) handleSharedUser(ctx context.Context, req ctrl.Request,
315	userID int64) (user *backendv1.User, currentStatus backendv1.UserStatus, stop bool, result ctrl.Result, err error) {	97✔
316	currentStatus.Backend, err = r.getSharedBackendURL(ctx, req.Namespace, userID)	97✔
317	if err != nil {	97✔
318	currentStatus.State = err.Error()	×
319	result = ctrl.Result{RequeueAfter: 2 * time.Second}	×
320	stop = true	×
321	err = nil	×
322	return	×
323	}	×
324
325	engineStatus, err := r.engineStatus(ctx, userID, currentStatus.Backend)	97✔
326	if err != nil {	105✔
327	currentStatus.State = err.Error()	8✔
328	if isDialError(err) {	16✔
329	stop = true	8✔
330	err = nil	8✔
331	result = ctrl.Result{RequeueAfter: 2 * time.Second}	8✔
332	return	8✔
333	}	8✔
334	return	×
335	}
336	currentStatus.State = string(engineStatus)	89✔
337		89✔
338	user = &backendv1.User{}	89✔
339	if err = r.Client.Get(ctx, req.NamespacedName, user); err != nil {	107✔
340	if apierrors.IsNotFound(err) {	36✔
341	// User is already deleted, kill engine if it's still running.	18✔
342	if engineStatus == Running \|\| engineStatus == Idle {	24✔
343	r.Log.Info("kill engine for already deleted user:", "user", userID)	6✔
344	err = r.killEngine(ctx, userID, currentStatus.Backend)	6✔
345	return	6✔
346	}	6✔
347	err = nil	12✔
348	stop = true	12✔
349	}
350	return	12✔
351	}
352
353	if !user.ObjectMeta.DeletionTimestamp.IsZero() {	71✔
354	// object is marked for deletion	×
355	if engineStatus == Running \|\| engineStatus == Idle {	×
356	r.Log.Info("kill engine for user marked for deletion:", "user", userID)	×
357	if err = r.killEngine(ctx, userID, currentStatus.Backend); err != nil {	×
358	r.Log.Error(err, "fail to kill engine marked for deletion")	×
359	}	×
360	}
361	stop = true	×
362	return	×
363	}
364	return	71✔
365	}
366
367	func (r *UserReconciler) getSharedBackendURL(ctx context.Context, namespace string, uid int64) (urlStr string, err error) {	97✔
368	endpoints := &corev1.Endpoints{}	97✔
369	// fetch backend endpoints.	97✔
370	if err = r.Get(ctx, client.ObjectKey{Namespace: namespace, Name: "shared-backend"}, endpoints); err != nil {	97✔
371	return	×
372	}	×
373	// StatefulSet endpoints are always ordered by name.
374	var backendURLs []string	97✔
375	for _, subset := range endpoints.Subsets {	194✔
376	for _, addr := range subset.Addresses {	194✔
377	u := fmt.Sprintf("http://%s:%d", addr.IP, subset.Ports[0].Port)	97✔
378	backendURLs = append(backendURLs, u)	97✔
379	}	97✔
380	}
381	if len(backendURLs) == 0 {	97✔
382	err = fmt.Errorf("backend endpoints not found")	×
383	} else {	97✔
384	urlStr = backendURLs[uid%int64(len(backendURLs))]	97✔
385	}	97✔
386	return	97✔
387	}
388
389	func (r *UserReconciler) createDeployment(ctx context.Context, name, namespace string) error {	5✔
390	deployment := NewDeployment(&r.developerDeploymentTemplate, name, namespace)	5✔
391	if err := r.Client.Create(ctx, deployment); err != nil {	5✔
392	return err	×
393	}	×
394	return nil	5✔
395	}
396
397	func (r *UserReconciler) createService(ctx context.Context, name, namespace string) error {	5✔
398	service := NewService(&r.developerServiceTemplate, name, namespace)	5✔
399	if err := r.Client.Create(ctx, service); err != nil {	5✔
400	return err	×
401	}	×
402	return nil	5✔
403	}
404
405	func (r *UserReconciler) getDBEntry(keyPrefix string, userID int64, fn dbQueryFunc, useCache bool) (interface{}, error) {	335✔
406	cacheKey := fmt.Sprintf("%s-%d", keyPrefix, userID)	335✔
407	cacheEntry, ok := r.localCache[cacheKey]	335✔
408	if useCache && ok && cacheEntry.Expiration.After(time.Now()) {	581✔
409	return cacheEntry.Value, nil	246✔
410	}	246✔
411	db := sql.GetDB()	89✔
412	v, err := fn(db, userID)	89✔
413	if err != nil {	89✔
414	return nil, err	×
415	}	×
416	r.localCache[cacheKey] = CacheEntry{v, time.Now().Add(kDefaultCacheDuration)}	89✔
417	return v, nil	89✔
418	}
419
420	func getUser(db *gorm.DB, userID int64) (interface{}, error) {	49✔
421	return sql.GetUser(db, userID)	49✔
422	}	49✔
423
424	func getDeveloperKey(db *gorm.DB, userID int64) (interface{}, error) {	40✔
425	return sql.GetDeveloperKey(db, userID)	40✔
426	}	40✔
427
428	func (r UserReconciler) httpWithContext(ctx context.Context, method, url string, body io.Reader) (http.Response, error) {	282✔
429	r.Log.Info(fmt.Sprintf("HTTP %v url:%v", method, url))	282✔
430	req, err := http.NewRequestWithContext(ctx, method, url, body)	282✔
431	if err != nil {	282✔
432	return nil, err	×
433	}	×
434	if body != nil {	322✔
435	req.Header.Set("Content-Type", "application/json")	40✔
436	}	40✔
437	resp, err := http.DefaultClient.Do(req)	282✔
438	return resp, err	282✔
439	}
440
441	func (r *UserReconciler) engineStatus(ctx context.Context, userID int64, userURL string) (UserState, error) {	225✔
442	resp, err := r.httpWithContext(ctx, "GET", fmt.Sprintf("%s/engine-status?userid=%d", userURL, userID), nil)	225✔
443	if err != nil {	272✔
444	return "", err	47✔
445	}	47✔
446	defer resp.Body.Close()	178✔
447	if resp.StatusCode != http.StatusOK {	178✔
448	return "", fmt.Errorf("Failed to get engine status: %+v", resp)	×
449	}	×
450	jsonResponse := &JSONResponse{Data: ""}	178✔
451	if err := json.NewDecoder(resp.Body).Decode(jsonResponse); err != nil {	178✔
452	return "", err	×
453	}	×
454	r.Log.Info("engine status:", "user", userID, "resp", jsonResponse)	178✔
455	return UserState(jsonResponse.Data.(string)), nil	178✔
456	}
457
458	func (r *UserReconciler) killEngine(ctx context.Context, userID int64, backendURL string) error {	17✔
459	resp, err := r.httpWithContext(ctx, "GET", fmt.Sprintf("%s/kill-engine?userid=%d", backendURL, userID), nil)	17✔
460	if err != nil {	18✔
461	return err	1✔
462	}	1✔
463	if resp.StatusCode != http.StatusOK {	16✔
464	return fmt.Errorf("Failed killEngine http status: %+v", resp)	×
465	}	×
466	return nil	16✔
467	}
468
469	func (r *UserReconciler) runEngine(ctx context.Context, userID int64, userURL string) error {	40✔
470	v, err := r.getDBEntry("user", userID, getUser, false)	40✔
471	if err != nil {	40✔
472	return err	×
473	}	×
474	u := v.(*sql.User)	40✔
475	v, err = r.getDBEntry("developer-key", userID, getDeveloperKey, false)	40✔
476	if err != nil {	40✔
477	return err	×
478	}	×
479	developerKey := v.(*string)	40✔
480		40✔
481	token, err := dbproxy.SignToken(u.ID)	40✔
482	if err != nil {	40✔
483	return err	×
484	}	×
485
486	options := platformOptions(u, developerKey, config.GetAlmondConfig().DatabaseProxyURL, token)	40✔
487		40✔
488	b, err := json.Marshal(options)	40✔
489	if err != nil {	40✔
490	return err	×
491	}	×
492	resp, err := r.httpWithContext(ctx, "POST", fmt.Sprintf("%s/run-engine", userURL), bytes.NewReader(b))	40✔
493	if err != nil {	40✔
494	return err	×
495	}	×
496	defer resp.Body.Close()	40✔
497	if resp.StatusCode != http.StatusOK {	40✔
498	return fmt.Errorf("Failed runEngine http status: %+v", resp)	×
499	}	×
500
501	return nil	40✔
502	}
503
504	func (r *UserReconciler) deleteDeploymentService(ctx context.Context, req ctrl.Request, userID int64) error {	5✔
505	r.Log.Info("Deleting deployment and service:", "user", userID)	5✔
506	meta := &metav1.ObjectMeta{Name: req.Name, Namespace: req.Namespace}	5✔
507	err1 := r.deleteService(ctx, req, userID, meta)	5✔
508	err2 := r.deleteDeployment(ctx, req, meta)	5✔
509	if err1 != nil && err2 != nil {	5✔
510	return fmt.Errorf("Error1:%v\nError2:%v", err1, err2)	×
511	}	×
512	if err1 != nil {	5✔
513	return err1	×
514	}	×
515	if err2 != nil {	5✔
516	return err2	×
517	}	×
518	return nil	5✔
519	}
520
521	func (r UserReconciler) deleteDeployment(ctx context.Context, req ctrl.Request, meta metav1.ObjectMeta) error {	5✔
522	deployment := &appsv1.Deployment{ObjectMeta: *meta}	5✔
523	if err := r.Client.Get(ctx, req.NamespacedName, deployment); err != nil {	6✔
524	if apierrors.IsNotFound(err) {	2✔
525	return nil	1✔
526	}	1✔
527	}
528	if err := r.Client.Delete(ctx, deployment); err != nil {	4✔
529	return err	×
530	}	×
531	return nil	4✔
532	}
533
534	func (r UserReconciler) deleteService(ctx context.Context, req ctrl.Request, userID int64, meta metav1.ObjectMeta) error {	5✔
535	service := &corev1.Service{ObjectMeta: *meta}	5✔
536	if err := r.Client.Get(ctx, req.NamespacedName, service); err != nil {	6✔
537	if apierrors.IsNotFound(err) {	2✔
538	return nil	1✔
539	}	1✔
540	}
541	if len(service.Spec.ClusterIP) > 0 && len(service.Spec.Ports) > 0 {	8✔
542	backend := fmt.Sprintf("http://%s:%d", service.Spec.ClusterIP, service.Spec.Ports[0].Port)	4✔
543	if err := r.killEngine(ctx, userID, backend); err != nil {	5✔
544	r.Log.Error(err, "kill engine failed")	1✔
545	}	1✔
546	}
547	if err := r.Client.Delete(ctx, service); err != nil {	4✔
548	return err	×
549	}	×
550	return nil	4✔
551	}
552
553	// SetupWithManager sets up the controller with the Manager.
554	func (r *UserReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
555	return ctrl.NewControllerManagedBy(mgr).	1✔
556	For(&backendv1.User{}).	1✔
557	Complete(r)	1✔
558	}	1✔

stanford-oval / almond-cloud / 3619

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous