mirror of
https://github.com/tuna/tunasync.git
synced 2025-04-20 11:42:43 +00:00
fix(tunasync): connection leakage caused by http keep-alive
This commit is contained in:
bigeagle
parent
d1981379a4
commit
2268eb3b0f
@ -8,6 +8,7 @@ import (
|
||||
"errors"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
// GetTLSConfig generate tls.Config from CAFile
|
||||
@ -28,20 +29,34 @@ func GetTLSConfig(CAFile string) (*tls.Config, error) {
|
||||
return tlsConfig, nil
|
||||
}
|
||||
|
||||
// PostJSON posts json object to url
|
||||
func PostJSON(url string, obj interface{}, tlsConfig *tls.Config) (*http.Response, error) {
|
||||
var client *http.Client
|
||||
if tlsConfig == nil {
|
||||
client = &http.Client{}
|
||||
} else {
|
||||
tr := &http.Transport{
|
||||
TLSClientConfig: tlsConfig,
|
||||
}
|
||||
client = &http.Client{
|
||||
Transport: tr,
|
||||
// CreateHTTPClient returns a http.Client
|
||||
func CreateHTTPClient(CAFile string) (*http.Client, error) {
|
||||
var tlsConfig *tls.Config
|
||||
var err error
|
||||
|
||||
if CAFile != "" {
|
||||
tlsConfig, err = GetTLSConfig(CAFile)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
tr := &http.Transport{
|
||||
MaxIdleConnsPerHost: 20,
|
||||
TLSClientConfig: tlsConfig,
|
||||
}
|
||||
|
||||
return &http.Client{
|
||||
Transport: tr,
|
||||
Timeout: 5 * time.Second,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// PostJSON posts json object to url
|
||||
func PostJSON(url string, obj interface{}, client *http.Client) (*http.Response, error) {
|
||||
if client == nil {
|
||||
client, _ = CreateHTTPClient("")
|
||||
}
|
||||
b := new(bytes.Buffer)
|
||||
if err := json.NewEncoder(b).Encode(obj); err != nil {
|
||||
return nil, err
|
||||
@ -50,17 +65,9 @@ func PostJSON(url string, obj interface{}, tlsConfig *tls.Config) (*http.Respons
|
||||
}
|
||||
|
||||
// GetJSON gets a json response from url
|
||||
func GetJSON(url string, obj interface{}, tlsConfig *tls.Config) (*http.Response, error) {
|
||||
var client *http.Client
|
||||
if tlsConfig == nil {
|
||||
client = &http.Client{}
|
||||
} else {
|
||||
tr := &http.Transport{
|
||||
TLSClientConfig: tlsConfig,
|
||||
}
|
||||
client = &http.Client{
|
||||
Transport: tr,
|
||||
}
|
||||
func GetJSON(url string, obj interface{}, client *http.Client) (*http.Response, error) {
|
||||
if client == nil {
|
||||
client, _ = CreateHTTPClient("")
|
||||
}
|
||||
|
||||
resp, err := client.Get(url)
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
package manager
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
@ -20,10 +19,10 @@ var manager *Manager
|
||||
|
||||
// A Manager represents a manager server
|
||||
type Manager struct {
|
||||
cfg *Config
|
||||
engine *gin.Engine
|
||||
adapter dbAdapter
|
||||
tlsConfig *tls.Config
|
||||
cfg *Config
|
||||
engine *gin.Engine
|
||||
adapter dbAdapter
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
// GetTUNASyncManager returns the manager from config
|
||||
@ -37,19 +36,18 @@ func GetTUNASyncManager(cfg *Config) *Manager {
|
||||
gin.SetMode(gin.ReleaseMode)
|
||||
}
|
||||
s := &Manager{
|
||||
cfg: cfg,
|
||||
engine: gin.Default(),
|
||||
adapter: nil,
|
||||
tlsConfig: nil,
|
||||
cfg: cfg,
|
||||
engine: gin.Default(),
|
||||
adapter: nil,
|
||||
}
|
||||
|
||||
if cfg.Files.CACert != "" {
|
||||
tlsConfig, err := GetTLSConfig(cfg.Files.CACert)
|
||||
httpClient, err := CreateHTTPClient(cfg.Files.CACert)
|
||||
if err != nil {
|
||||
logger.Error("Error initializing TLS config: %s", err.Error())
|
||||
logger.Error("Error initializing HTTP client: %s", err.Error())
|
||||
return nil
|
||||
}
|
||||
s.tlsConfig = tlsConfig
|
||||
s.httpClient = httpClient
|
||||
}
|
||||
|
||||
if cfg.Files.DBFile != "" {
|
||||
@ -96,12 +94,20 @@ func (s *Manager) setDBAdapter(adapter dbAdapter) {
|
||||
// Run runs the manager server forever
|
||||
func (s *Manager) Run() {
|
||||
addr := fmt.Sprintf("%s:%d", s.cfg.Server.Addr, s.cfg.Server.Port)
|
||||
|
||||
httpServer := &http.Server{
|
||||
Addr: addr,
|
||||
Handler: s.engine,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 10 * time.Second,
|
||||
}
|
||||
|
||||
if s.cfg.Server.SSLCert == "" && s.cfg.Server.SSLKey == "" {
|
||||
if err := s.engine.Run(addr); err != nil {
|
||||
if err := httpServer.ListenAndServe(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
} else {
|
||||
if err := s.engine.RunTLS(addr, s.cfg.Server.SSLCert, s.cfg.Server.SSLKey); err != nil {
|
||||
if err := httpServer.ListenAndServeTLS(s.cfg.Server.SSLCert, s.cfg.Server.SSLKey); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
@ -258,7 +264,7 @@ func (s *Manager) handleClientCmd(c *gin.Context) {
|
||||
}
|
||||
|
||||
// post command to worker
|
||||
_, err = PostJSON(workerURL, workerCmd, s.tlsConfig)
|
||||
_, err = PostJSON(workerURL, workerCmd, s.httpClient)
|
||||
if err != nil {
|
||||
err := fmt.Errorf("post command to worker %s(%s) fail: %s", workerID, workerURL, err.Error())
|
||||
c.Error(err)
|
||||
|
||||
@ -2,7 +2,6 @@ package worker
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"html/template"
|
||||
@ -26,8 +25,8 @@ type Worker struct {
|
||||
semaphore chan empty
|
||||
|
||||
schedule *scheduleQueue
|
||||
httpServer *gin.Engine
|
||||
tlsConfig *tls.Config
|
||||
httpEngine *gin.Engine
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
// GetTUNASyncWorker returns a singalton worker
|
||||
@ -48,12 +47,12 @@ func GetTUNASyncWorker(cfg *Config) *Worker {
|
||||
}
|
||||
|
||||
if cfg.Manager.CACert != "" {
|
||||
tlsConfig, err := GetTLSConfig(cfg.Manager.CACert)
|
||||
httpClient, err := CreateHTTPClient(cfg.Manager.CACert)
|
||||
if err != nil {
|
||||
logger.Error("Failed to init TLS config: %s", err.Error())
|
||||
logger.Error("Error initializing HTTP client: %s", err.Error())
|
||||
return nil
|
||||
}
|
||||
w.tlsConfig = tlsConfig
|
||||
w.httpClient = httpClient
|
||||
}
|
||||
|
||||
w.initJobs()
|
||||
@ -227,18 +226,25 @@ func (w *Worker) makeHTTPServer() {
|
||||
c.JSON(http.StatusOK, gin.H{"msg": "OK"})
|
||||
})
|
||||
|
||||
w.httpServer = s
|
||||
w.httpEngine = s
|
||||
}
|
||||
|
||||
func (w *Worker) runHTTPServer() {
|
||||
addr := fmt.Sprintf("%s:%d", w.cfg.Server.Addr, w.cfg.Server.Port)
|
||||
|
||||
httpServer := &http.Server{
|
||||
Addr: addr,
|
||||
Handler: w.httpEngine,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 10 * time.Second,
|
||||
}
|
||||
|
||||
if w.cfg.Server.SSLCert == "" && w.cfg.Server.SSLKey == "" {
|
||||
if err := w.httpServer.Run(addr); err != nil {
|
||||
if err := httpServer.ListenAndServe(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
} else {
|
||||
if err := w.httpServer.RunTLS(addr, w.cfg.Server.SSLCert, w.cfg.Server.SSLKey); err != nil {
|
||||
if err := httpServer.ListenAndServeTLS(w.cfg.Server.SSLCert, w.cfg.Server.SSLKey); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
@ -345,7 +351,7 @@ func (w *Worker) registorWorker() {
|
||||
URL: w.URL(),
|
||||
}
|
||||
|
||||
if _, err := PostJSON(url, msg, w.tlsConfig); err != nil {
|
||||
if _, err := PostJSON(url, msg, w.httpClient); err != nil {
|
||||
logger.Error("Failed to register worker")
|
||||
}
|
||||
}
|
||||
@ -368,7 +374,7 @@ func (w *Worker) updateStatus(jobMsg jobMessage) {
|
||||
ErrorMsg: jobMsg.msg,
|
||||
}
|
||||
|
||||
if _, err := PostJSON(url, smsg, w.tlsConfig); err != nil {
|
||||
if _, err := PostJSON(url, smsg, w.httpClient); err != nil {
|
||||
logger.Error("Failed to update mirror(%s) status: %s", jobMsg.name, err.Error())
|
||||
}
|
||||
}
|
||||
@ -382,7 +388,7 @@ func (w *Worker) fetchJobStatus() []MirrorStatus {
|
||||
w.Name(),
|
||||
)
|
||||
|
||||
if _, err := GetJSON(url, &mirrorList, w.tlsConfig); err != nil {
|
||||
if _, err := GetJSON(url, &mirrorList, w.httpClient); err != nil {
|
||||
logger.Error("Failed to fetch job status: %s", err.Error())
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user