Support queueing updates by type and profile

CHANGELOG.md

@@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 ## [unreleased]
 ### Added
+- [#4982](https://github.com/apache/trafficcontrol/issues/4982) Added the ability to support queueing updates by server type and profile 
 - [#5412](https://github.com/apache/trafficcontrol/issues/5412) Added last authenticated time to user API's (`GET /user/current, GET /users, GET /user?id=`) response payload
 - [#5451](https://github.com/apache/trafficcontrol/issues/5451) Added change log count to user API's response payload and query param (username) to logs API
 - Added support for CDN locks

docs/source/api/v4/cdns_id_queue_update.rst

@@ -37,12 +37,22 @@ Request Structure
 	| ID   | The integral, unique identifier for the CDN on which to (de)queue updates |
 	+------+---------------------------------------------------------------------------+
 
+.. table:: Request Query Parameters
+
+	+-----------+----------+---------------------------------------------------------------------------------------------------------------+
+	| Name      | Required | Description                                                                                                   |
+	+===========+==========+===============================================================================================================+
+	| type      | no       | The name of the ``type`` of servers, for which the updates need to be queued or dequeued.                     |
+	+-----------+----------+---------------------------------------------------------------------------------------------------------------+
+	| profile   | no       | The name of the ``profile`` of servers, for which the updates need to be queued or dequeued.                  |
+	+-----------+----------+---------------------------------------------------------------------------------------------------------------+
+
 :action: One of "queue" or "dequeue" as appropriate
 
 .. code-block:: http
 	:caption: Request Example
 
-	POST /api/4.0/cdns/2/queue_update HTTP/1.1
+	POST /api/4.0/cdns/2/queue_update?type=EDGE HTTP/1.1
 	Host: trafficops.infra.ciab.test
 	User-Agent: curl/7.47.0
 	Accept: */*

traffic_ops/testing/api/v4/cdn_queue_updates_by_type_profile_test.go

@@ -0,0 +1,215 @@
+package v4
+
+/*
+
+   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.
+*/
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+
+	client "github.com/apache/trafficcontrol/traffic_ops/v4-client"
+)
+
+func TestCDNQueueUpdateByProfileAndType(t *testing.T) {
+	WithObjs(t, []TCObj{Types, CDNs, Profiles, Statuses, Divisions, Regions, PhysLocations, CacheGroups, Servers}, func() {
+		QueueUpdatesByType(t)
+		QueueUpdatesByProfile(t)
+	})
+}
+
+func QueueUpdatesByType(t *testing.T) {
+	allServersResp, _, err := TOSession.GetServers(client.NewRequestOptions())
+	if err != nil {
+		t.Fatalf("couldn't get all servers: %v", err)
+	}
+
+	// Clear updates on all servers to begin with
+	for _, s := range allServersResp.Response {
+		if s.ID != nil {
+			_, _, err = TOSession.SetServerQueueUpdate(*s.ID, false, client.NewRequestOptions())
+			if err != nil {
+				t.Errorf("couldn't clear updates on server with ID: %d, err: %v", *s.ID, err.Error())
+			}
+		}
+	}
+
+	allServersResp, _, err = TOSession.GetServers(client.NewRequestOptions())
+	if err != nil {
+		t.Fatalf("couldn't get all servers: %v", err)
+	}
+	queryOpts := client.NewRequestOptions()
+	if len(testData.Servers) < 1 {
+		t.Fatalf("no servers to run the tests on...quitting.")
+	}
+	server := testData.Servers[0]
+	opts := client.NewRequestOptions()
+	if server.CDNName == nil {
+		t.Fatalf("server doesn't have a CDN name...quitting")
+	}
+	opts.QueryParameters.Set("name", *server.CDNName)
+
+	// Get the first server's CDN ID
+	cdns, _, err := TOSession.GetCDNs(opts)
+	if err != nil {
+		t.Fatalf("error while getting CDNs: %v", err)
+	}
+	if len(cdns.Response) < 1 {
+		t.Fatalf("expected 1 CDN in response, got %d", len(cdns.Response))
+	}
+	opts.QueryParameters.Del("name")
+	queryOpts.QueryParameters.Set("type", server.Type)
+	// Queue updates by type (and CDN)
+	_, _, err = TOSession.QueueUpdatesForCDN(cdns.Response[0].ID, true, queryOpts)
+	if err != nil {
+		t.Errorf("couldn't queue updates by type (and CDN): %v", err)
+	}
+
+	// Get all the servers for the same CDN and type as that of the first server
+	opts.QueryParameters.Set("cdn", strconv.Itoa(cdns.Response[0].ID))
+	opts.QueryParameters.Set("type", server.Type)
+	serverIDMap := make(map[int]bool, 0)
+	resp, _, err := TOSession.GetServers(opts)
+	if err != nil {
+		t.Fatalf("couldn't get servers by cdn and type: %v", err)
+	}
+	if len(resp.Response) < 1 {
+		t.Fatalf("expected atleast one server in response, got %d", len(resp.Response))
+	}
+	for _, s := range resp.Response {
+		if s.UpdPending == nil || !*s.UpdPending {
+			t.Errorf("expected updates to be queued on all the servers filtered by type and CDN, but %s didn't queue updates", *s.HostName)
+		}
+		if s.ID != nil {
+			serverIDMap[*s.ID] = true
+		}
+	}
+
+	// Make sure that the servers that are not filtered by the above criteria do not have updates queued
+	allServersResp, _, err = TOSession.GetServers(client.NewRequestOptions())
+	if err != nil {
+		t.Fatalf("couldn't get all servers: %v", err)
+	}
+	for _, s := range allServersResp.Response {
+		if s.ID != nil {
+			if _, ok := serverIDMap[*s.ID]; !ok {
+				if s.UpdPending != nil && *s.UpdPending {
+					t.Errorf("did not expect server with ID: %d to have queued updates", *s.ID)
+				}
+			}
+
+		}
+	}
+	_, _, err = TOSession.QueueUpdatesForCDN(cdns.Response[0].ID, false, queryOpts)
+	if err != nil {
+		t.Errorf("couldn't queue updates by type (and CDN): %v", err)
+	}
+}
+
+func QueueUpdatesByProfile(t *testing.T) {
+	allServersResp, _, err := TOSession.GetServers(client.NewRequestOptions())
+	if err != nil {
+		t.Fatalf("couldn't get all servers: %v", err)
+	}
+
+	// Clear updates on all servers to begin with
+	for _, s := range allServersResp.Response {
+		if s.ID != nil {
+			_, _, err = TOSession.SetServerQueueUpdate(*s.ID, false, client.NewRequestOptions())
+			if err != nil {
+				t.Errorf("couldn't clear updates on server with ID: %d, err: %v", *s.ID, err.Error())
+			}
+		}
+	}
+
+	queryOpts := client.NewRequestOptions()
+	if len(testData.Servers) < 1 {
+		t.Fatalf("no servers to run the tests on...quitting.")
+	}
+	server := testData.Servers[0]
+	opts := client.NewRequestOptions()
+	if server.CDNName == nil || server.Profile == nil {
+		t.Fatalf("server doesn't have a CDN name or a profile name...quitting")
+	}
+
+	//Get the first server's CDN ID
+	opts.QueryParameters.Set("name", strings.TrimSpace(*server.CDNName))
+
+	cdns, _, err := TOSession.GetCDNs(opts)
+	if err != nil {
+		t.Fatalf("error while getting CDNs: %v", err)
+	}
+	if len(cdns.Response) < 1 {
+		t.Fatalf("expected 1 CDN in response, got %d", len(cdns.Response))
+	}
+	opts.QueryParameters.Del("name")
+
+	// Get the first server's Profile ID
+	opts.QueryParameters.Set("name", *server.Profile)
+	profiles, _, err := TOSession.GetProfiles(opts)
+	if err != nil {
+		t.Fatalf("error while getting profiles: %v", err)
+	}
+	if len(profiles.Response) < 1 {
+		t.Fatalf("expected 1 profile in response, got %d", len(profiles.Response))
+	}
+	opts.QueryParameters.Del("name")
+	queryOpts.QueryParameters.Set("profile", profiles.Response[0].Name)
+	// Queue updates by profile (and CDN)
+	_, _, err = TOSession.QueueUpdatesForCDN(cdns.Response[0].ID, true, queryOpts)
+	if err != nil {
+		t.Errorf("couldn't queue updates by profile (and CDN): %v", err)
+	}
+
+	// Get all the servers for the same CDN and profile as that of the first server
+	opts.QueryParameters.Set("cdn", strconv.Itoa(cdns.Response[0].ID))
+	opts.QueryParameters.Set("profileId", strconv.Itoa(profiles.Response[0].ID))
+	serverIDMap := make(map[int]bool, 0)
+	resp, _, err := TOSession.GetServers(opts)
+	if err != nil {
+		t.Fatalf("couldn't get servers by cdn and profile: %v", err)
+	}
+	if len(resp.Response) < 1 {
+		t.Fatalf("expected atleast one server in response, got %d", len(resp.Response))
+	}
+	for _, s := range resp.Response {
+		if s.UpdPending == nil || !*s.UpdPending {
+			t.Errorf("expected updates to be queued on all the servers filtered by profile and CDN, but %s didn't queue updates", *s.HostName)
+		}
+		if s.ID != nil {
+			serverIDMap[*s.ID] = true
+		}
+	}
+
+	// Make sure that the servers that are not filtered by the above criteria do not have updates queued
+	allServersResp, _, err = TOSession.GetServers(client.NewRequestOptions())
+	if err != nil {
+		t.Fatalf("couldn't get all servers: %v", err)
+	}
+	for _, s := range allServersResp.Response {
+		if s.ID != nil {
+			if _, ok := serverIDMap[*s.ID]; !ok {
+				if s.UpdPending != nil && *s.UpdPending {
+					t.Errorf("did not expect server with ID: %d to have queued updates", *s.ID)
+				}
+			}
+
+		}
+	}
+	_, _, err = TOSession.QueueUpdatesForCDN(cdns.Response[0].ID, false, queryOpts)
+	if err != nil {
+		t.Errorf("couldn't queue updates by type (and CDN): %v", err)
+	}
+}

traffic_ops/traffic_ops_golang/cdn/queue.go

@@ -20,25 +20,44 @@ package cdn
  */
 
 import (
-	"database/sql"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 	"strconv"
 
 	"github.com/apache/trafficcontrol/lib/go-tc"
-
+	"github.com/apache/trafficcontrol/lib/go-util"
 	"github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/api"
 	"github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/dbhelpers"
+
+	"github.com/jmoiron/sqlx"
 )
 
 func Queue(w http.ResponseWriter, r *http.Request) {
+	var typeID int
+	var profileID int
+	var ok bool
+	var err error
+	var str string
+	params := make(map[string]string, 0)
+
 	inf, userErr, sysErr, errCode := api.NewInfo(r, []string{"id"}, []string{"id"})
 	if userErr != nil || sysErr != nil {
 		api.HandleErr(w, r, inf.Tx.Tx, errCode, userErr, sysErr)
 		return
 	}
 	defer inf.Close()
+
+	cols := map[string]dbhelpers.WhereColumnInfo{
+		"cdnID":     {Column: "server.cdn_id", Checker: api.IsInt},
+		"typeID":    {Column: "server.type", Checker: nil},
+		"profileID": {Column: "server.profile", Checker: nil},
+	}
+
+	typeName := inf.Params["type"]
+	profile := inf.Params["profile"]
+
 	reqObj := tc.CDNQueueUpdateRequest{}
 	if err := json.NewDecoder(r.Body).Decode(&reqObj); err != nil {
 		api.HandleErr(w, r, inf.Tx.Tx, http.StatusBadRequest, errors.New("malformed JSON: "+err.Error()), nil)
@@ -48,6 +67,7 @@ func Queue(w http.ResponseWriter, r *http.Request) {
 		api.HandleErr(w, r, inf.Tx.Tx, http.StatusBadRequest, errors.New("action must be 'queue' or 'dequeue'"), nil)
 		return
 	}
+	params["cdnID"] = strconv.Itoa(inf.IntParams["id"])
 	cdnName, ok, err := dbhelpers.GetCDNNameFromID(inf.Tx.Tx, int64(inf.IntParams["id"]))
 	if err != nil {
 		api.HandleErr(w, r, inf.Tx.Tx, http.StatusInternalServerError, nil, errors.New("getting cdn name from ID '"+inf.Params["id"]+"': "+err.Error()))
@@ -56,22 +76,72 @@ func Queue(w http.ResponseWriter, r *http.Request) {
 		api.HandleErr(w, r, inf.Tx.Tx, http.StatusNotFound, nil, nil)
 		return
 	}
+
+	// get type ID
+	if typeName != "" {
+		typeID, ok, err = dbhelpers.GetTypeIDByName(typeName, inf.Tx.Tx)
+		if err != nil {
+			api.HandleErr(w, r, inf.Tx.Tx, http.StatusInternalServerError, nil, errors.New("error getting type ID from name: "+err.Error()))
+			return
+		}
+		if !ok {
+			api.HandleErr(w, r, inf.Tx.Tx, http.StatusNotFound, errors.New("no type ID found with that name"), nil)
+			return
+		}
+		params["typeID"] = strconv.Itoa(typeID)
+		str = fmt.Sprintf(" typeID: %d", typeID)
+	}
+
+	// get profile ID
+	if profile != "" {
+		profileID, ok, err = dbhelpers.GetProfileIDFromName(profile, inf.Tx.Tx)
+		if err != nil {
+			api.HandleErr(w, r, inf.Tx.Tx, http.StatusInternalServerError, nil, errors.New("error getting profile ID from name: "+err.Error()))
+			return
+		}
+		if !ok {
+			api.HandleErr(w, r, inf.Tx.Tx, http.StatusNotFound, errors.New("no profile ID found with that name"), nil)
+			return
+		}
+		params["profileID"] = strconv.Itoa(profileID)
+		str = fmt.Sprintf(" profileID: %d", profileID)
+	}
+
 	userErr, sysErr, statusCode := dbhelpers.CheckIfCurrentUserHasCdnLock(inf.Tx.Tx, string(cdnName), inf.User.UserName)
 	if userErr != nil || sysErr != nil {
 		api.HandleErr(w, r, inf.Tx.Tx, statusCode, userErr, sysErr)
 		return
 	}
-	if err := queueUpdates(inf.Tx.Tx, int64(inf.IntParams["id"]), reqObj.Action == "queue"); err != nil {
-		api.HandleErr(w, r, inf.Tx.Tx, http.StatusInternalServerError, nil, errors.New("CDN queueing updates: "+err.Error()))
+
+	where, orderBy, pagination, queryValues, errs := dbhelpers.BuildWhereAndOrderByAndPagination(params, cols)
+	if len(errs) > 0 {
+		errCode = http.StatusBadRequest
+		userErr = util.JoinErrs(errs)
+		api.HandleErr(w, r, inf.Tx.Tx, errCode, userErr, nil)
 		return
 	}
-	api.CreateChangeLogRawTx(api.ApiChange, "CDN: "+string(cdnName)+", ID: "+strconv.Itoa(inf.IntParams["id"])+", ACTION: CDN server updates "+reqObj.Action+"d", inf.User, inf.Tx.Tx)
+
+	query := `UPDATE server SET upd_pending = :upd_pending`
+	query = query + where + orderBy + pagination
+	queryValues["upd_pending"] = reqObj.Action == "queue"
+	rowsAffected, err := queueUpdates(inf.Tx, queryValues, query)
+	if err != nil {
+		api.HandleErr(w, r, inf.Tx.Tx, http.StatusInternalServerError, nil, fmt.Errorf("queueing updates: %v", err))
+		return
+	}
+
+	api.CreateChangeLogRawTx(api.ApiChange, "CDN: "+string(cdnName)+", ID: "+strconv.Itoa(inf.IntParams["id"])+str+", ACTION: server updates "+reqObj.Action+"d on "+strconv.Itoa(int(rowsAffected))+" servers", inf.User, inf.Tx.Tx)
 	api.WriteResp(w, r, tc.CDNQueueUpdateResponse{Action: reqObj.Action, CDNID: int64(inf.IntParams["id"])})
 }
 
-func queueUpdates(tx *sql.Tx, cdnID int64, queue bool) error {
-	if _, err := tx.Exec(`UPDATE server SET upd_pending = $1 WHERE server.cdn_id = $2`, queue, cdnID); err != nil {
-		return errors.New("querying queue updates: " + err.Error())
+// queueUpdates is the helper function to queue/ dequeue updates on servers for a CDN, optionally filtered by type and/ or profile
+func queueUpdates(tx *sqlx.Tx, queryValues map[string]interface{}, query string) (int64, error) {
+	result, err := tx.NamedExec(query, queryValues)
+	if err != nil {
+		return 0, errors.New("querying queue updates: " + err.Error())
+	} else if rc, err := result.RowsAffected(); err != nil {
+		return rc, fmt.Errorf("checking rows updated: %v", err)
+	} else {
+		return rc, nil
 	}
-	return nil
 }

traffic_ops/v4-client/cdn.go

@@ -63,3 +63,16 @@ func (to *Session) GetCDNSSLKeys(name string, opts RequestOptions) (tc.CDNSSLKey
 	reqInf, err := to.get(route, opts, &data)
 	return data, reqInf, err
 }
+
+// QueueUpdatesForCDN set the "updPending" field of a list of servers identified by
+// 'cdnID' and any other query params (type or profile) to the value of 'queueUpdate'
+func (to *Session) QueueUpdatesForCDN(cdnID int, queueUpdate bool, opts RequestOptions) (tc.CDNQueueUpdateResponse, toclientlib.ReqInf, error) {
+	req := tc.CDNQueueUpdateRequest{Action: queueUpdateActions[queueUpdate]}
+	var resp tc.CDNQueueUpdateResponse
+	if opts.QueryParameters == nil {
+		opts.QueryParameters = url.Values{}
+	}
+	path := fmt.Sprintf("/cdns/%d/queue_update", cdnID)
+	reqInf, err := to.post(path, opts, req, &resp)
+	return resp, reqInf, err
+}