Skip to content

Commit

Permalink
refactor server update behavior
Browse files Browse the repository at this point in the history
  • Loading branch information
@harshavardhana
harshavardhana committed Jul 23, 2020
1 parent b9be841 commit a9ba680
Show file tree
Hide file tree
Showing 10 changed files with 284 additions and 342 deletions.
105 changes: 58 additions & 47 deletions cmd/admin-handlers.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,31 +66,13 @@ const (
mgmtForceStop = "forceStop"
)

func updateServer(updateURL, sha256Hex string, latestReleaseTime time.Time) (us madmin.ServerUpdateStatus, err error) {
minioMode := getMinioMode()
// No inputs provided we should try to update using the default URL.
if updateURL == "" && sha256Hex == "" && latestReleaseTime.IsZero() {
var updateMsg string
updateMsg, sha256Hex, _, latestReleaseTime, err = getUpdateInfo(updateTimeout, minioMode)
if err != nil {
return us, err
}
if updateMsg == "" {
us.CurrentVersion = Version
us.UpdatedVersion = Version
return us, nil
}
if runtime.GOOS == "windows" {
updateURL = minioReleaseURL + "minio.exe"
} else {
updateURL = minioReleaseURL + "minio"
}
}
if err = doUpdate(updateURL, sha256Hex, minioMode); err != nil {
func updateServer(u *url.URL, sha256Sum []byte, lrTime time.Time, mode string) (us madmin.ServerUpdateStatus, err error) {
if err = doUpdate(u, lrTime, sha256Sum, mode); err != nil {
return us, err
}

us.CurrentVersion = Version
us.UpdatedVersion = latestReleaseTime.Format(minioReleaseTagTimeLayout)
us.UpdatedVersion = lrTime.Format(minioReleaseTagTimeLayout)
return us, nil
}

Expand All @@ -115,47 +97,71 @@ func (a adminAPIHandlers) ServerUpdateHandler(w http.ResponseWriter, r *http.Req
}

vars := mux.Vars(r)
updateURL := vars[peerRESTUpdateURL]
updateURL := vars["updateURL"]
mode := getMinioMode()
var sha256Hex string
var latestReleaseTime time.Time
if updateURL != "" {
u, err := url.Parse(updateURL)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
if updateURL == "" {
updateURL = minioReleaseInfoURL
if runtime.GOOS == globalWindowsOSName {
updateURL = minioReleaseWindowsInfoURL
}
}

content, err := downloadReleaseURL(updateURL, updateTimeout, mode)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
u, err := url.Parse(updateURL)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}

content, err := downloadReleaseURL(u, updateTimeout, mode)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}

sha256Sum, lrTime, err := parseReleaseData(content)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}

u.Path = path.Dir(u.Path) + SlashSeparator + "minio.RELEASE." + lrTime.Format(minioReleaseTagTimeLayout)

crTime, err := GetCurrentReleaseTime()
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}

if lrTime.Sub(crTime) < 0 {
updateStatus := madmin.ServerUpdateStatus{
CurrentVersion: Version,
UpdatedVersion: Version,
}

sha256Hex, latestReleaseTime, err = parseReleaseData(content)
// Marshal API response
jsonBytes, err := json.Marshal(updateStatus)
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}

if runtime.GOOS == "windows" {
u.Path = path.Dir(u.Path) + SlashSeparator + "minio.exe"
} else {
u.Path = path.Dir(u.Path) + SlashSeparator + "minio"
}

updateURL = u.String()
writeSuccessResponseJSON(w, jsonBytes)
return
}

for _, nerr := range globalNotificationSys.ServerUpdate(updateURL, sha256Hex, latestReleaseTime) {
for _, nerr := range globalNotificationSys.ServerUpdate(ctx, u, sha256Sum, lrTime) {
if nerr.Err != nil {
logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
logger.LogIf(ctx, nerr.Err)
err = fmt.Errorf("Server update failed, please do not restart the servers yet: failed with %w", nerr.Err)
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
}

updateStatus, err := updateServer(updateURL, sha256Hex, latestReleaseTime)
updateStatus, err := updateServer(u, sha256Sum, lrTime, mode)
if err != nil {
err = fmt.Errorf("Server update failed, please do not restart the servers yet: failed with %w", err)
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
}
Expand All @@ -169,10 +175,15 @@ func (a adminAPIHandlers) ServerUpdateHandler(w http.ResponseWriter, r *http.Req

writeSuccessResponseJSON(w, jsonBytes)

if updateStatus.CurrentVersion != updateStatus.UpdatedVersion {
// We did upgrade - restart all services.
globalServiceSignalCh <- serviceRestart
// Notify all other MinIO peers signal service.
for _, nerr := range globalNotificationSys.SignalService(serviceRestart) {
if nerr.Err != nil {
logger.GetReqInfo(ctx).SetTags("peerAddress", nerr.Host.String())
logger.LogIf(ctx, nerr.Err)
}
}

globalServiceSignalCh <- serviceRestart
}

// ServiceHandler - POST /minio/admin/v3/service?action={action}
Expand Down
47 changes: 38 additions & 9 deletions cmd/common-main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,10 @@ import (
"encoding/gob"
"errors"
"net"
"net/url"
"os"
"path/filepath"
"runtime"
"strings"
"time"

Expand Down Expand Up @@ -68,16 +70,43 @@ func verifyObjectLayerFeatures(name string, objAPI ObjectLayer) {

// Check for updates and print a notification message
func checkUpdate(mode string) {
updateURL := minioReleaseInfoURL
if runtime.GOOS == globalWindowsOSName {
updateURL = minioReleaseWindowsInfoURL
}

u, err := url.Parse(updateURL)
if err != nil {
return
}

// Its OK to ignore any errors during doUpdate() here.
if updateMsg, _, currentReleaseTime, latestReleaseTime, err := getUpdateInfo(2*time.Second, mode); err == nil {
if updateMsg == "" {
return
}
if globalInplaceUpdateDisabled {
logStartupMessage(updateMsg)
} else {
logStartupMessage(prepareUpdateMessage("Run `mc admin update`", latestReleaseTime.Sub(currentReleaseTime)))
}
crTime, err := GetCurrentReleaseTime()
if err != nil {
return
}

_, lrTime, err := getLatestReleaseTime(u, 2*time.Second, mode)
if err != nil {
return
}

var older time.Duration
var downloadURL string
if lrTime.After(crTime) {
older = lrTime.Sub(crTime)
downloadURL = getDownloadURL(releaseTimeToReleaseTag(lrTime))
}

updateMsg := prepareUpdateMessage(downloadURL, older)
if updateMsg == "" {
return
}

if globalInplaceUpdateDisabled {
logStartupMessage(updateMsg)
} else {
logStartupMessage(prepareUpdateMessage("Run `mc admin update`", lrTime.Sub(crTime)))
}
}

Expand Down
6 changes: 3 additions & 3 deletions cmd/notification.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -401,15 +401,15 @@ func (sys *NotificationSys) DownloadProfilingData(ctx context.Context, writer io
}

// ServerUpdate - updates remote peers.
func (sys *NotificationSys) ServerUpdate(updateURL, sha256Hex string, latestReleaseTime time.Time) []NotificationPeerErr {
func (sys *NotificationSys) ServerUpdate(ctx context.Context, u *url.URL, sha256Sum []byte, lrTime time.Time) []NotificationPeerErr {
ng := WithNPeers(len(sys.peerClients))
for idx, client := range sys.peerClients {
if client == nil {
continue
}
client := client
ng.Go(GlobalContext, func() error {
return client.ServerUpdate(updateURL, sha256Hex, latestReleaseTime)
ng.Go(ctx, func() error {
return client.ServerUpdate(ctx, u, sha256Sum, lrTime)
}, idx, *client.host)
}
return ng.Wait()
Expand Down
23 changes: 15 additions & 8 deletions cmd/peer-rest-client.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -616,17 +616,24 @@ func (client *peerRESTClient) LoadGroup(group string) error {
return nil
}

type serverUpdateInfo struct {
URL *url.URL
Sha256Sum []byte
Time time.Time
}

// ServerUpdate - sends server update message to remote peers.
func (client *peerRESTClient) ServerUpdate(updateURL, sha256Hex string, latestReleaseTime time.Time) error {
func (client *peerRESTClient) ServerUpdate(ctx context.Context, u *url.URL, sha256Sum []byte, lrTime time.Time) error {
values := make(url.Values)
values.Set(peerRESTUpdateURL, updateURL)
values.Set(peerRESTSha256Hex, sha256Hex)
if !latestReleaseTime.IsZero() {
values.Set(peerRESTLatestRelease, latestReleaseTime.Format(time.RFC3339))
} else {
values.Set(peerRESTLatestRelease, "")
var reader bytes.Buffer
if err := gob.NewEncoder(&reader).Encode(serverUpdateInfo{
URL: u,
Sha256Sum: sha256Sum,
Time: lrTime,
}); err != nil {
return err
}
respBody, err := client.call(peerRESTMethodServerUpdate, values, nil, -1)
respBody, err := client.callWithContext(ctx, peerRESTMethodServerUpdate, values, &reader, -1)
if err != nil {
return err
}
Expand Down
29 changes: 13 additions & 16 deletions cmd/peer-rest-common.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
package cmd

const (
peerRESTVersion = "v9"
peerRESTVersion = "v10"
peerRESTVersionPrefix = SlashSeparator + peerRESTVersion
peerRESTPrefix = minioReservedBucketPath + "/peer"
peerRESTPath = peerRESTPrefix + peerRESTVersionPrefix
Expand Down Expand Up @@ -59,21 +59,18 @@ const (
)

const (
peerRESTBucket = "bucket"
peerRESTUser = "user"
peerRESTGroup = "group"
peerRESTUserTemp = "user-temp"
peerRESTPolicy = "policy"
peerRESTUserOrGroup = "user-or-group"
peerRESTIsGroup = "is-group"
peerRESTUpdateURL = "updateURL"
peerRESTSha256Hex = "sha256Hex"
peerRESTLatestRelease = "latestReleaseTime"
peerRESTSignal = "signal"
peerRESTProfiler = "profiler"
peerRESTDryRun = "dry-run"
peerRESTTraceAll = "all"
peerRESTTraceErr = "err"
peerRESTBucket = "bucket"
peerRESTUser = "user"
peerRESTGroup = "group"
peerRESTUserTemp = "user-temp"
peerRESTPolicy = "policy"
peerRESTUserOrGroup = "user-or-group"
peerRESTIsGroup = "is-group"
peerRESTSignal = "signal"
peerRESTProfiler = "profiler"
peerRESTDryRun = "dry-run"
peerRESTTraceAll = "all"
peerRESTTraceErr = "err"

peerRESTListenBucket = "bucket"
peerRESTListenPrefix = "prefix"
Expand Down
27 changes: 11 additions & 16 deletions cmd/peer-rest-server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -773,25 +773,21 @@ func (s *peerRESTServer) ServerUpdateHandler(w http.ResponseWriter, r *http.Requ
return
}

vars := mux.Vars(r)
updateURL := vars[peerRESTUpdateURL]
sha256Hex := vars[peerRESTSha256Hex]
var latestReleaseTime time.Time
var err error
if latestRelease := vars[peerRESTLatestRelease]; latestRelease != "" {
latestReleaseTime, err = time.Parse(time.RFC3339, latestRelease)
if err != nil {
s.writeErrorResponse(w, err)
return
}
if r.ContentLength < 0 {
s.writeErrorResponse(w, errInvalidArgument)
return
}
us, err := updateServer(updateURL, sha256Hex, latestReleaseTime)

var info serverUpdateInfo
err := gob.NewDecoder(r.Body).Decode(&info)
if err != nil {
s.writeErrorResponse(w, err)
return
}
if us.CurrentVersion != us.UpdatedVersion {
globalServiceSignalCh <- serviceRestart

if _, err = updateServer(info.URL, info.Sha256Sum, info.Time, getMinioMode()); err != nil {
s.writeErrorResponse(w, err)
return
}
}

Expand Down Expand Up @@ -1046,8 +1042,7 @@ func registerPeerRESTHandlers(router *mux.Router) {
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodDeleteBucketMetadata).HandlerFunc(httpTraceHdrs(server.DeleteBucketMetadataHandler)).Queries(restQueries(peerRESTBucket)...)
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodLoadBucketMetadata).HandlerFunc(httpTraceHdrs(server.LoadBucketMetadataHandler)).Queries(restQueries(peerRESTBucket)...)
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodSignalService).HandlerFunc(httpTraceHdrs(server.SignalServiceHandler)).Queries(restQueries(peerRESTSignal)...)
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodServerUpdate).HandlerFunc(httpTraceHdrs(server.ServerUpdateHandler)).Queries(restQueries(peerRESTUpdateURL, peerRESTSha256Hex, peerRESTLatestRelease)...)

subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodServerUpdate).HandlerFunc(httpTraceHdrs(server.ServerUpdateHandler))
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodDeletePolicy).HandlerFunc(httpTraceAll(server.DeletePolicyHandler)).Queries(restQueries(peerRESTPolicy)...)
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodLoadPolicy).HandlerFunc(httpTraceAll(server.LoadPolicyHandler)).Queries(restQueries(peerRESTPolicy)...)
subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodLoadPolicyMapping).HandlerFunc(httpTraceAll(server.LoadPolicyMappingHandler)).Queries(restQueries(peerRESTUserOrGroup)...)
Expand Down
Loading

0 comments on commit a9ba680

Please sign in to comment.