3615

pending completion

Build # 3615

Build Type

push

travis-ci-com

Committed by

jgd5

Commit Message

Backend: hot fix user-controller

Run Details

2143 of 3909 branches covered (54.82%)

Branch coverage included in aggregate %.

1 of 2 new or added lines in 1 file covered. (50.0%)

8 existing lines in 1 file now uncovered.

8243 of 11915 relevant lines covered (69.18%)

583.29 hits per line

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

74.02

/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
                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 {	39✔
85	return &PlatformOptions{	39✔
86	UserID: user.ID,	39✔
87	CloudID: user.CloudID,	39✔
88	AuthToken: user.AuthToken,	39✔
89	DeveloperKey: developerKey,	39✔
90	Locale: user.Locale,	39✔
91	Timezone: user.Timezone,	39✔
92	DBProxyURL: dbProxyURL,	39✔
93	DBProxyAccessToken: dbProxyToken,	39✔
94	HumanName: user.HumanName,	39✔
95	Email: user.Email,	39✔
96	}	39✔
97	}	39✔
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) {	249✔
139	_ = r.Log.WithValues("user-controller", req.NamespacedName)	249✔
140	r.Log.Info("--- start ---")	249✔
141	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)	249✔
142	defer cancel()	249✔
143		249✔
144	var (	249✔
145	err error	249✔
146	userID int64	249✔
147	trustedDeveloper bool	249✔
148	user *backendv1.User	249✔
149	stop bool	249✔
150	result ctrl.Result	249✔
151	currentStatus backendv1.UserStatus	249✔
152	)	249✔
153		249✔
154	defer func() {	498✔
155	if user != nil {	490✔
156	if len(user.Spec.Mode) == 0 {	283✔
157	if r.almondConfig.EnableDeveloperBackend && trustedDeveloper {	52✔
158	user.Spec.Mode = "developer"	10✔
159	} else {	42✔
160	user.Spec.Mode = "shared"	32✔
161	}	32✔
162	r.Update(ctx, user)	42✔
163	}
164	user.Status = currentStatus	241✔
165	r.Status().Update(ctx, user)	241✔
166	r.Log.Info("update status:", "user", user.Spec.ID, "status", user.Status)	241✔
167	}
168	r.Log.Info("--- end ---")	249✔
169	}()
170
171	userID, trustedDeveloper, err = r.getUserFromName(req.Name)	249✔
172	if err != nil {	249✔
173	return ctrl.Result{}, err	×
174	}	×
175
176	if r.almondConfig.EnableDeveloperBackend && trustedDeveloper {	402✔
177	user, currentStatus, stop, result, err = r.handleDeveloper(ctx, req, userID)	153✔
178	} else {	249✔
179	user, currentStatus, stop, result, err = r.handleSharedUser(ctx, req, userID)	96✔
180	}	96✔
181	if err != nil \|\| stop {	332✔
182	return result, err	83✔
183	}	83✔
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 {	286✔
194	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil	120✔
195	}	120✔
196
197	if UserState(currentStatus.State) == Idle {	53✔
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 {	39✔
UNCOV 206	currentStatus.State = err.Error()	×
UNCOV 207	return ctrl.Result{RequeueAfter: 2 * time.Second}, nil	×
UNCOV 208	}	×
209	currentStatus.State = string(Starting)	39✔
210	return ctrl.Result{RequeueAfter: 1 * time.Second}, nil	39✔
211	}
212
213	func (r *UserReconciler) getUserFromName(name string) (uid int64, trustedDeveloper bool, err error) {	249✔
214	uid, err = strconv.ParseInt(strings.TrimPrefix(name, "user-"), 10, 64)	249✔
215	if err != nil {	249✔
216	return	×
217	}	×
218	v, err := r.getDBEntry("user", uid, getUser, true)	249✔
219	if err != nil {	249✔
220	return	×
221	}	×
222	user := v.(*sql.User)	249✔
223	if user.Roles&kTrustedDeveloper != 0 {	402✔
224	trustedDeveloper = true	153✔
225	}	153✔
226	return	249✔
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) {	153✔
231	user = &backendv1.User{}	153✔
232	// for developer users, make sure deployment and service are up before proceeding.	153✔
233	if err = r.Client.Get(ctx, req.NamespacedName, user); err != nil {	158✔
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	return	5✔
241	}
242	if !user.ObjectMeta.DeletionTimestamp.IsZero() {	148✔
243	// user is marked for deletion	×
244	currentStatus.State = string(Stopping)	×
245	if err = r.deleteDeploymentService(ctx, req, userID); err != nil {	×
246	r.Log.Error(err, "fail to delete developer deployment or service marked for deletion")	×
247	}	×
NEW 248	stop = true	×
UNCOV 249	return	×
250	}
251	deployment := &appsv1.Deployment{}	148✔
252	if err = r.Client.Get(ctx, req.NamespacedName, deployment); err != nil {	153✔
253	if apierrors.IsNotFound(err) {	10✔
254	if err = r.createDeployment(ctx, req.Name, req.Namespace); err != nil {	5✔
255	return	×
256	}	×
257	stop = true	5✔
258	result = ctrl.Result{RequeueAfter: 2 * time.Second}	5✔
259	// setting err to nil so ControllerManager will respect the RequeueAfter time	5✔
260	err = nil	5✔
261	return	5✔
262	}
263	return	×
264	}
265	if deployment.Status.AvailableReplicas <= 0 {	154✔
266	currentStatus.State = string(Starting)	11✔
267	stop = true	11✔
268	result = ctrl.Result{RequeueAfter: 2 * time.Second}	11✔
269	return	11✔
270	}	11✔
271	service := &corev1.Service{}	132✔
272	if err = r.Client.Get(ctx, req.NamespacedName, service); err != nil {	137✔
273	if apierrors.IsNotFound(err) {	10✔
274	if err = r.createService(ctx, req.Name, req.Namespace); err != nil {	5✔
275	return	×
276	}	×
277	currentStatus.State = string(Starting)	5✔
278	stop = true	5✔
279	result = ctrl.Result{RequeueAfter: 2 * time.Second}	5✔
280	err = nil	5✔
281	return	5✔
282	}
283	return	×
284	}
285	if len(service.Spec.ClusterIP) == 0 {	127✔
286	currentStatus.State = string(Starting)	×
287	stop = true	×
288	result = ctrl.Result{RequeueAfter: 2 * time.Second}	×
289	err = nil	×
290	return	×
291	}	×
292	if len(service.Spec.Ports) == 0 {	127✔
293	err = errors.New("service port not configured")	×
294	return	×
295	}	×
296	currentStatus.Backend = fmt.Sprintf("http://%s:%d", service.Spec.ClusterIP, service.Spec.Ports[0].Port)	127✔
297		127✔
298	engineStatus, err := r.engineStatus(ctx, userID, currentStatus.Backend)	127✔
299	if err != nil {	164✔
300	currentStatus.State = err.Error()	37✔
301	if isDialError(err) {	74✔
302	stop = true	37✔
303	err = nil	37✔
304	result = ctrl.Result{RequeueAfter: 2 * time.Second}	37✔
305	return	37✔
306	}	37✔
307	return	×
308	}
309	currentStatus.State = string(engineStatus)	90✔
310	return	90✔
311	}
312
313	func (r *UserReconciler) handleSharedUser(ctx context.Context, req ctrl.Request,
314	userID int64) (user *backendv1.User, currentStatus backendv1.UserStatus, stop bool, result ctrl.Result, err error) {	96✔
315	currentStatus.Backend, err = r.getSharedBackendURL(ctx, req.Namespace, userID)	96✔
316	if err != nil {	96✔
317	currentStatus.State = err.Error()	×
318	result = ctrl.Result{RequeueAfter: 2 * time.Second}	×
319	stop = true	×
320	err = nil	×
321	return	×
322	}	×
323
324	engineStatus, err := r.engineStatus(ctx, userID, currentStatus.Backend)	96✔
325	if err != nil {	104✔
326	currentStatus.State = err.Error()	8✔
327	if isDialError(err) {	16✔
328	stop = true	8✔
329	err = nil	8✔
330	result = ctrl.Result{RequeueAfter: 2 * time.Second}	8✔
331	return	8✔
332	}	8✔
333	return	×
334	}
335	currentStatus.State = string(engineStatus)	88✔
336		88✔
337	user = &backendv1.User{}	88✔
338	if err = r.Client.Get(ctx, req.NamespacedName, user); err != nil {	106✔
339	if apierrors.IsNotFound(err) {	36✔
340	// User is already deleted, kill engine if it's still running.	18✔
341	if engineStatus == Running \|\| engineStatus == Idle {	24✔
342	r.Log.Info("kill engine for already deleted user:", "user", userID)	6✔
343	err = r.killEngine(ctx, userID, currentStatus.Backend)	6✔
344	return	6✔
345	}	6✔
346	err = nil	12✔
347	stop = true	12✔
348	}
349	return	12✔
350	}
351
352	if !user.ObjectMeta.DeletionTimestamp.IsZero() {	70✔
353	// object is marked for deletion	×
354	if engineStatus == Running \|\| engineStatus == Idle {	×
355	r.Log.Info("kill engine for user marked for deletion:", "user", userID)	×
356	if err = r.killEngine(ctx, userID, currentStatus.Backend); err != nil {	×
357	r.Log.Error(err, "fail to kill engine marked for deletion")	×
358	}	×
359	}
360	stop = true	×
361	return	×
362	}
363	return	70✔
364	}
365
366	func (r *UserReconciler) getSharedBackendURL(ctx context.Context, namespace string, uid int64) (urlStr string, err error) {	96✔
367	endpoints := &corev1.Endpoints{}	96✔
368	// fetch backend endpoints.	96✔
369	if err = r.Get(ctx, client.ObjectKey{Namespace: namespace, Name: "shared-backend"}, endpoints); err != nil {	96✔
370	return	×
371	}	×
372	// StatefulSet endpoints are always ordered by name.
373	var backendURLs []string	96✔
374	for _, subset := range endpoints.Subsets {	192✔
375	for _, addr := range subset.Addresses {	192✔
376	u := fmt.Sprintf("http://%s:%d", addr.IP, subset.Ports[0].Port)	96✔
377	backendURLs = append(backendURLs, u)	96✔
378	}	96✔
379	}
380	if len(backendURLs) == 0 {	96✔
381	err = fmt.Errorf("backend endpoints not found")	×
382	} else {	96✔
383	urlStr = backendURLs[uid%int64(len(backendURLs))]	96✔
384	}	96✔
385	return	96✔
386	}
387
388	func (r *UserReconciler) createDeployment(ctx context.Context, name, namespace string) error {	5✔
389	deployment := NewDeployment(&r.developerDeploymentTemplate, name, namespace)	5✔
390	if err := r.Client.Create(ctx, deployment); err != nil {	5✔
391	return err	×
392	}	×
393	return nil	5✔
394	}
395
396	func (r *UserReconciler) createService(ctx context.Context, name, namespace string) error {	5✔
397	service := NewService(&r.developerServiceTemplate, name, namespace)	5✔
398	if err := r.Client.Create(ctx, service); err != nil {	5✔
399	return err	×
400	}	×
401	return nil	5✔
402	}
403
404	func (r *UserReconciler) getDBEntry(keyPrefix string, userID int64, fn dbQueryFunc, useCache bool) (interface{}, error) {	327✔
405	cacheKey := fmt.Sprintf("%s-%d", keyPrefix, userID)	327✔
406	cacheEntry, ok := r.localCache[cacheKey]	327✔
407	if useCache && ok && cacheEntry.Expiration.After(time.Now()) {	567✔
408	return cacheEntry.Value, nil	240✔
409	}	240✔
410	db := sql.GetDB()	87✔
411	v, err := fn(db, userID)	87✔
412	if err != nil {	87✔
UNCOV 413	return nil, err	×
UNCOV 414	}	×
415	r.localCache[cacheKey] = CacheEntry{v, time.Now().Add(kDefaultCacheDuration)}	87✔
416	return v, nil	87✔
417	}
418
419	func getUser(db *gorm.DB, userID int64) (interface{}, error) {	48✔
420	return sql.GetUser(db, userID)	48✔
421	}	48✔
422
423	func getDeveloperKey(db *gorm.DB, userID int64) (interface{}, error) {	39✔
424	return sql.GetDeveloperKey(db, userID)	39✔
425	}	39✔
426
427	func (r UserReconciler) httpWithContext(ctx context.Context, method, url string, body io.Reader) (http.Response, error) {	279✔
428	r.Log.Info(fmt.Sprintf("HTTP %v url:%v", method, url))	279✔
429	req, err := http.NewRequestWithContext(ctx, method, url, body)	279✔
430	if err != nil {	279✔
431	return nil, err	×
432	}	×
433	if body != nil {	318✔
434	req.Header.Set("Content-Type", "application/json")	39✔
435	}	39✔
436	resp, err := http.DefaultClient.Do(req)	279✔
437	return resp, err	279✔
438	}
439
440	func (r *UserReconciler) engineStatus(ctx context.Context, userID int64, userURL string) (UserState, error) {	223✔
441	resp, err := r.httpWithContext(ctx, "GET", fmt.Sprintf("%s/engine-status?userid=%d", userURL, userID), nil)	223✔
442	if err != nil {	268✔
443	return "", err	45✔
444	}	45✔
445	defer resp.Body.Close()	178✔
446	if resp.StatusCode != http.StatusOK {	178✔
447	return "", fmt.Errorf("Failed to get engine status: %+v", resp)	×
448	}	×
449	jsonResponse := &JSONResponse{Data: ""}	178✔
450	if err := json.NewDecoder(resp.Body).Decode(jsonResponse); err != nil {	178✔
451	return "", err	×
452	}	×
453	r.Log.Info("engine status:", "user", userID, "resp", jsonResponse)	178✔
454	return UserState(jsonResponse.Data.(string)), nil	178✔
455	}
456
457	func (r *UserReconciler) killEngine(ctx context.Context, userID int64, backendURL string) error {	17✔
458	resp, err := r.httpWithContext(ctx, "GET", fmt.Sprintf("%s/kill-engine?userid=%d", backendURL, userID), nil)	17✔
459	if err != nil {	18✔
460	return err	1✔
461	}	1✔
462	if resp.StatusCode != http.StatusOK {	16✔
463	return fmt.Errorf("Failed killEngine http status: %+v", resp)	×
464	}	×
465	return nil	16✔
466	}
467
468	func (r *UserReconciler) runEngine(ctx context.Context, userID int64, userURL string) error {	39✔
469	v, err := r.getDBEntry("user", userID, getUser, false)	39✔
470	if err != nil {	39✔
UNCOV 471	return err	×
UNCOV 472	}	×
473	u := v.(*sql.User)	39✔
474	v, err = r.getDBEntry("developer-key", userID, getDeveloperKey, false)	39✔
475	if err != nil {	39✔
476	return err	×
477	}	×
478	developerKey := v.(*string)	39✔
479		39✔
480	token, err := dbproxy.SignToken(u.ID)	39✔
481	if err != nil {	39✔
482	return err	×
483	}	×
484
485	options := platformOptions(u, developerKey, config.GetAlmondConfig().DatabaseProxyURL, token)	39✔
486		39✔
487	b, err := json.Marshal(options)	39✔
488	if err != nil {	39✔
489	return err	×
490	}	×
491	resp, err := r.httpWithContext(ctx, "POST", fmt.Sprintf("%s/run-engine", userURL), bytes.NewReader(b))	39✔
492	if err != nil {	39✔
493	return err	×
494	}	×
495	defer resp.Body.Close()	39✔
496	if resp.StatusCode != http.StatusOK {	39✔
497	return fmt.Errorf("Failed runEngine http status: %+v", resp)	×
498	}	×
499
500	return nil	39✔
501	}
502
503	func (r *UserReconciler) deleteDeploymentService(ctx context.Context, req ctrl.Request, userID int64) error {	5✔
504	r.Log.Info("Deleting deployment and service:", "user", userID)	5✔
505	meta := &metav1.ObjectMeta{Name: req.Name, Namespace: req.Namespace}	5✔
506	err1 := r.deleteService(ctx, req, userID, meta)	5✔
507	err2 := r.deleteDeployment(ctx, req, meta)	5✔
508	if err1 != nil && err2 != nil {	5✔
509	return fmt.Errorf("Error1:%v\nError2:%v", err1, err2)	×
510	}	×
511	if err1 != nil {	5✔
512	return err1	×
513	}	×
514	if err2 != nil {	5✔
515	return err2	×
516	}	×
517	return nil	5✔
518	}
519
520	func (r UserReconciler) deleteDeployment(ctx context.Context, req ctrl.Request, meta metav1.ObjectMeta) error {	5✔
521	deployment := &appsv1.Deployment{ObjectMeta: *meta}	5✔
522	if err := r.Client.Get(ctx, req.NamespacedName, deployment); err != nil {	6✔
523	if apierrors.IsNotFound(err) {	2✔
524	return nil	1✔
525	}	1✔
526	}
527	if err := r.Client.Delete(ctx, deployment); err != nil {	4✔
528	return err	×
529	}	×
530	return nil	4✔
531	}
532
533	func (r UserReconciler) deleteService(ctx context.Context, req ctrl.Request, userID int64, meta metav1.ObjectMeta) error {	5✔
534	service := &corev1.Service{ObjectMeta: *meta}	5✔
535	if err := r.Client.Get(ctx, req.NamespacedName, service); err != nil {	6✔
536	if apierrors.IsNotFound(err) {	2✔
537	return nil	1✔
538	}	1✔
539	}
540	if len(service.Spec.ClusterIP) > 0 && len(service.Spec.Ports) > 0 {	8✔
541	backend := fmt.Sprintf("http://%s:%d", service.Spec.ClusterIP, service.Spec.Ports[0].Port)	4✔
542	if err := r.killEngine(ctx, userID, backend); err != nil {	5✔
543	r.Log.Error(err, "kill engine failed")	1✔
544	}	1✔
545	}
546	if err := r.Client.Delete(ctx, service); err != nil {	4✔
547	return err	×
548	}	×
549	return nil	4✔
550	}
551
552	// SetupWithManager sets up the controller with the Manager.
553	func (r *UserReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
554	return ctrl.NewControllerManagedBy(mgr).	1✔
555	For(&backendv1.User{}).	1✔
556	Complete(r)	1✔
557	}	1✔

stanford-oval / almond-cloud / 3615

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