Skip to content

Commit

Permalink
Merge pull request PelicanPlatform#1673 from h2zh/persist-server-down…
Browse files Browse the repository at this point in the history
…time

Persist server downtime
  • Loading branch information
jhiemstrawisc authored Nov 12, 2024
2 parents 48e8576 + 075cdec commit c2a3caa
Show file tree
Hide file tree
Showing 19 changed files with 592 additions and 16 deletions.
1 change: 1 addition & 0 deletions cmd/fed_serve_cache_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ func TestFedServeCache(t *testing.T) {
viper.Set("Registry.RequireOriginApproval", false)
viper.Set("Registry.RequireCacheApproval", false)
viper.Set("Origin.EnablePublicReads", false)
viper.Set("Director.DbLocation", filepath.Join(t.TempDir(), "director.sqlite"))

require.NoError(t, err)

Expand Down
2 changes: 2 additions & 0 deletions cmd/fed_serve_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ func TestFedServePosixOrigin(t *testing.T) {
viper.Set("Registry.DbLocation", filepath.Join(t.TempDir(), "ns-registry.sqlite"))
viper.Set("Registry.RequireOriginApproval", false)
viper.Set("Registry.RequireCacheApproval", false)
viper.Set("Director.DbLocation", filepath.Join(t.TempDir(), "director.sqlite"))

defer cancel()

_, fedCancel, err := launchers.LaunchModules(ctx, modules)
Expand Down
1 change: 1 addition & 0 deletions cmd/plugin_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,7 @@ func (f *FedTest) Spinup() {
viper.Set("Origin.Port", 0)
viper.Set("Server.WebPort", 0)
viper.Set("Origin.RunLocation", tmpPath)
viper.Set("Director.DbLocation", filepath.Join(f.T.TempDir(), "director.sqlite"))

err = config.InitServer(ctx, modules)
require.NoError(f.T, err)
Expand Down
2 changes: 2 additions & 0 deletions config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -1030,6 +1030,7 @@ func InitServer(ctx context.Context, currentServers server_structs.ServerType) e

viper.SetDefault("Origin.Multiuser", true)
viper.SetDefault(param.Origin_DbLocation.GetName(), "/var/lib/pelican/origin.sqlite")
viper.SetDefault(param.Director_DbLocation.GetName(), "/var/lib/pelican/director.sqlite")
viper.SetDefault("Director.GeoIPLocation", "/var/cache/pelican/maxmind/GeoLite2-City.mmdb")
viper.SetDefault("Registry.DbLocation", "/var/lib/pelican/registry.sqlite")
// The lotman db will actually take this path and create the lot at /path/.lot/lotman_cpp.sqlite
Expand All @@ -1040,6 +1041,7 @@ func InitServer(ctx context.Context, currentServers server_structs.ServerType) e
viper.SetDefault(param.Origin_GlobusConfigLocation.GetName(), filepath.Join("/run", "pelican", "xrootd", "origin", "globus"))
} else {
viper.SetDefault(param.Origin_DbLocation.GetName(), filepath.Join(configDir, "origin.sqlite"))
viper.SetDefault(param.Director_DbLocation.GetName(), filepath.Join(configDir, "director.sqlite"))
viper.SetDefault("Director.GeoIPLocation", filepath.Join(configDir, "maxmind", "GeoLite2-City.mmdb"))
viper.SetDefault("Registry.DbLocation", filepath.Join(configDir, "ns-registry.sqlite"))
// Lotdb will live at <configDir>/.lot/lotman_cpp.sqlite
Expand Down
24 changes: 18 additions & 6 deletions director/director_api.go
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ func listAdvertisement(serverTypes []server_structs.ServerType) []*server_struct
func checkFilter(serverName string) (bool, filterType) {
filteredServersMutex.RLock()
defer filteredServersMutex.RUnlock()

log.Debugf("Checking for a downtime filter applied to server %s", serverName)
status, exists := filteredServers[serverName]
// No filter entry
if !exists {
Expand Down Expand Up @@ -223,17 +223,29 @@ func hookServerAdsCache() {
})
}

// Populate internal filteredServers map by Director.FilteredServers
// Populate internal filteredServers map using Director.FilteredServers param and director db
func ConfigFilterdServers() {
filteredServersMutex.Lock()
defer filteredServersMutex.Unlock()

if !param.Director_FilteredServers.IsSet() {
return
if param.Director_FilteredServers.IsSet() {
for _, sn := range param.Director_FilteredServers.GetStringSlice() {
filteredServers[sn] = permFiltered
}
log.Debugln("Loaded server downtime configuration from the Director.FilteredServers parameter:", filteredServers)
}

for _, sn := range param.Director_FilteredServers.GetStringSlice() {
filteredServers[sn] = permFiltered
if param.Director_DbLocation.GetString() != "" {
persistedServerDowntimes, err := getAllServerDowntimes()
if err != nil {
log.Error("Failed to read persisted server downtimes from director db:", err)
return
}
for _, serverDowntime := range persistedServerDowntimes {
filteredServers[serverDowntime.Name] = serverDowntime.FilterType
}
log.Debugln("Loaded filtered servers config from director db:", filteredServers)
// if a filtered server config rule is set in both Director.FilteredServers param and director db, the latter one will eventually be used
}
}

Expand Down
157 changes: 157 additions & 0 deletions director/director_db.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
/***************************************************************
*
* Copyright (C) 2024, Pelican Project, Morgridge Institute for Research
*
* 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 director

import (
"embed"
"time"

"github.com/google/uuid"
"github.com/pkg/errors"
"gorm.io/gorm"
"gorm.io/gorm/logger"

"github.com/pelicanplatform/pelican/param"
"github.com/pelicanplatform/pelican/server_utils"
)

type ServerDowntime struct {
UUID string `gorm:"primaryKey"`
Name string `gorm:"not null;unique"`
FilterType filterType `gorm:"type:text;not null"`
// We don't use gorm default gorm.Model to change ID type to string
CreatedAt time.Time
UpdatedAt time.Time
}

var db *gorm.DB

//go:embed migrations/*.sql
var embedMigrations embed.FS

// Initialize the Director's sqlite database, which is used to persist information about server downtimes
func InitializeDB() error {
dbPath := param.Director_DbLocation.GetString()
tdb, err := server_utils.InitSQLiteDB(dbPath)
if err != nil {
return errors.Wrap(err, "failed to initialize the Director's sqlite database")
}
db = tdb
sqldb, err := db.DB()
if err != nil {
return errors.Wrapf(err, "failed to get sql.DB from gorm DB: %s", dbPath)
}
// Run database migrations
if err := server_utils.MigrateDB(sqldb, embedMigrations); err != nil {
return errors.Wrap(err, "failed to migrate the Director's sqlite database using embedded migration files")
}
return nil
}

// Shut down the Director's sqlite database
func shutdownDirectorDB() error {
return server_utils.ShutdownDB(db)
}

// Create a new db entry representing the downtime info of a server
func createServerDowntime(serverName string, filterType filterType) error {
id, err := uuid.NewV7()
if err != nil {
return errors.Wrap(err, "unable to create new UUID for new entry in server status table")
}
serverDowntime := ServerDowntime{
UUID: id.String(),
Name: serverName,
FilterType: filterType,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}

if err := db.Create(serverDowntime).Error; err != nil {
return errors.Wrap(err, "unable to create server downtime table")
}
return nil
}

// Retrieve the downtime info of a given server (filter applied to the server)
func getServerDowntime(serverName string) (filterType, error) {
var serverDowntime ServerDowntime
err := db.First(&serverDowntime, "name = ?", serverName).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return "", errors.Wrapf(err, "%s is not found in the Director db", serverName)
}
return "", errors.Wrapf(err, "unable to get the downtime of %s", serverName)
}
return filterType(serverDowntime.FilterType), nil
}

// Retrieve the downtime info of all servers saved in the Director's sqlite database
func getAllServerDowntimes() ([]ServerDowntime, error) {
var statuses []ServerDowntime
result := db.Find(&statuses)

if result.Error != nil {
return nil, errors.Wrap(result.Error, "unable to get the downtime of all servers")
}
return statuses, nil
}

// Set the downtime info (filterType) of a given server
func setServerDowntime(serverName string, filterType filterType) error {
var serverDowntime ServerDowntime
// slience the logger for this query because there's definitely an ErrRecordNotFound when a new downtime info entry inserted
err := db.Session(&gorm.Session{Logger: db.Logger.LogMode(logger.Silent)}).First(&serverDowntime, "name = ?", serverName).Error

// If the server doesn't exist in director db, create a new entry for it
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return createServerDowntime(serverName, filterType)
}

return errors.Wrapf(err, "unable to retrieve downtime status for server %s", serverName)
}

serverDowntime.FilterType = filterType
serverDowntime.UpdatedAt = time.Now()

if err := db.Save(&serverDowntime).Error; err != nil {
return errors.Wrap(err, "unable to update")
}
return nil
}

// Define a function type for setServerDowntime
type setServerDowntimeFunc func(string, filterType) error

// Make the function a variable so it can be mocked in tests
var setServerDowntimeFn setServerDowntimeFunc = setServerDowntime

// Delete the downtime info of a given server from the Director's sqlite database
func deleteServerDowntime(serverName string) error {
if err := db.Where("name = ?", serverName).Delete(&ServerDowntime{}).Error; err != nil {
return errors.Wrap(err, "failed to delete an entry in Server Status table")
}
return nil
}

// Define a function type for deleteServerDowntime
type deleteServerDowntimeFunc func(string) error

// Make the function a variable so it can be mocked in tests
var deleteServerDowntimeFn deleteServerDowntimeFunc = deleteServerDowntime
82 changes: 82 additions & 0 deletions director/director_db_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
package director

import (
"testing"

"github.com/glebarez/sqlite"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"

"github.com/pelicanplatform/pelican/server_utils"
)

var (
mockSS []ServerDowntime = []ServerDowntime{
{UUID: uuid.NewString(), Name: "/4a334d532d69:8443", FilterType: tempAllowed},
{UUID: uuid.NewString(), Name: "/my-origin.com/foo/Bar", FilterType: permFiltered},
{UUID: uuid.NewString(), Name: "/my-cache.com/chtc", FilterType: permFiltered},
}
)

func SetupMockDirectorDB(t *testing.T) {
mockDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
db = mockDB
require.NoError(t, err, "Error setting up mock origin DB")
err = db.AutoMigrate(&ServerDowntime{})
require.NoError(t, err, "Failed to migrate DB for Globus table")
}

func TeardownMockDirectorDB(t *testing.T) {
err := shutdownDirectorDB()
require.NoError(t, err, "Error tearing down mock director DB")
}

func insertMockDBData(ss []ServerDowntime) error {
return db.Create(&ss).Error
}

func TestDirectorDBBasics(t *testing.T) {
server_utils.ResetTestState()
SetupMockDirectorDB(t)
t.Cleanup(func() {
TeardownMockDirectorDB(t)
})
err := insertMockDBData(mockSS)
require.NoError(t, err)

t.Run("get-downtime", func(t *testing.T) {
filterType, err := getServerDowntime(mockSS[1].Name)
assert.Equal(t, filterType, permFiltered)
require.NoError(t, err)
})

t.Run("get-all-downtime", func(t *testing.T) {
statuses, err := getAllServerDowntimes()
require.NoError(t, err)
assert.Len(t, statuses, len(mockSS))
})

t.Run("set-downtime", func(t *testing.T) {
err = setServerDowntime(mockSS[1].Name, tempAllowed)
require.NoError(t, err)
filterType, err := getServerDowntime(mockSS[1].Name)
assert.Equal(t, filterType, tempAllowed)
require.NoError(t, err)
})

t.Run("duplicate-name-insert", func(t *testing.T) {
err := createServerDowntime(mockSS[1].Name, tempAllowed)
require.Error(t, err)
assert.Contains(t, err.Error(), "UNIQUE constraint failed")
})

t.Run("delete-downtime-entry-from-directory-db", func(t *testing.T) {
err = deleteServerDowntime(mockSS[0].Name)
require.NoError(t, err, "Error deleting server status")

_, err = getServerDowntime(mockSS[0].Name)
assert.Error(t, err, "Expected error retrieving deleted server status")
})
}
Loading

0 comments on commit c2a3caa

Please sign in to comment.