ayo/houston
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

로그 파일을 압축하지 않고 하나씩 업로드

Browse Source
This commit is contained in:
mountain
2024-11-06 16:46:45 +09:00
parent 9632dc7755
commit 99e1007012
4 changed files with 136 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
client/operation.go
View File

@ -15,11 +15,14 @@ import (
"path"
"path/filepath"
"regexp"
"slices"
"strconv"
"strings"
"syscall"
"time"
"github.com/djherbis/times"
"github.com/Knetic/govaluate"
"repositories.action2quare.com/ayo/gocommon/logger"
"repositories.action2quare.com/ayo/gocommon/metric"
@ -45,6 +48,47 @@ func lastExecutionArgs(verpath string) []string {
var errUploadZipLogFailed = errors.New("not ok")
func (hc *houstonClient) uploadRawLogFile(logFile string, name string, version string) error {
t, err := times.Stat(logFile)
if err != nil {
return err
}
file, err := os.Open(logFile)
if err != nil {
return err
}
if file == nil {
return errors.New("uploadRawLogFile failed : " + logFile)
}
defer file.Close()
req, err := http.NewRequest("POST", hc.config.HttpAddress+"/upload", file)
if err != nil {
logger.Println(err)
}
hn, _ := os.Hostname()
// createTime := file.
req.Header.Set("Houston-Service-Name", name)
req.Header.Set("Houston-Service-Version", version)
req.Header.Set("Houston-Service-Filename", t.BirthTime().UTC().Format(time.DateOnly)+"."+hn+path.Ext(logFile))
req.Header.Set("Content-Type", "application/zip")
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return errUploadZipLogFailed
}
return nil
}
func (hc *houstonClient) uploadZipLogFile(zipFile string, name string, version string) error {
zf, err := os.Open(zipFile)
if err != nil {
@ -99,6 +143,7 @@ func findMatchFiles(storageRoot, name, version, filter string) (string, []string
out = append(out, file)
}
slices.Sort(out)
return root, out
}
@ -318,33 +363,59 @@ func (hc *houstonClient) launch(meta *procmeta) error {
hc.siblingProcIndex[key] = runningFlags
}()
metricExporter := metric.NewPrometheusExport(hc.config.MetricNamespace)
defer func() {
reco := recover()
if reco != nil {
logger.Println(reco)
}
r.Close()
}()
metricExporter := metric.NewPrometheusExport(hc.config.MetricNamespace)
defer metricExporter.Shutdown()
total := 0
hn, _ := os.Hostname()
var targetFile *os.File
var currentFilePath string
ext := path.Ext(logfilePath)
head := logfilePath[:len(logfilePath)-len(ext)]
if len(head) > 0 && !strings.HasSuffix(head, "/") {
head += "."
}
reader := bufio.NewReader(r)
readingMetric := false
ext = "." + hn + ext
defer func() {
if targetFile != nil {
targetFile.Close()
targetFile = nil
hc.uploadRawLogFile(currentFilePath, meta.name, meta.version)
}
metricExporter.Shutdown()
}()
var metricBuffer []byte
wipeLogFile := func() {
total = 0
if targetFile != nil {
targetFile.Close()
targetFile = nil
hc.uploadRawLogFile(currentFilePath, meta.name, meta.version)
}
}
currentTime := time.Now().UTC()
for {
now := time.Now().UTC()
if now.YearDay() != currentTime.YearDay() {
wipeLogFile()
}
if targetFile == nil {
currentFile := head + time.Now().UTC().Format(".2006-01-02.150405") + ext
targetFile, _ = os.OpenFile(currentFile, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
currentTime = time.Now().UTC()
currentFilePath = head + currentTime.Format("2006-01-02.150405") + ext
targetFile, _ = os.OpenFile(currentFilePath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
}
buff, err := reader.ReadBytes('\n')
@ -402,12 +473,8 @@ func (hc *houstonClient) launch(meta *procmeta) error {
}
total += len(buff)
if total > 1024*1024 {
total = 0
targetFile.Close()
targetFile = nil
hc.uploadProcFiles(meta, "logs/*"+ext, true)
if total > 1024 {
wipeLogFile()
}
}
}
@ -423,18 +490,44 @@ func (hc *houstonClient) launch(meta *procmeta) error {
defer r.Close()
total := 0
hn, _ := os.Hostname()
var targetFile *os.File
var currentFilePath string
ext := path.Ext(logfilePath)
head := logfilePath[:len(logfilePath)-len(ext)]
if len(head) > 0 {
head += "."
}
reader := bufio.NewReader(r)
ext = "." + hn + ext
defer func() {
if targetFile != nil {
targetFile.Close()
targetFile = nil
}
hc.uploadProcFiles(meta, "logs/*"+ext, true)
}()
wipeLogFile := func() {
total = 0
if targetFile != nil {
targetFile.Close()
targetFile = nil
hc.uploadRawLogFile(currentFilePath, meta.name, meta.version)
}
}
currentTime := time.Now().UTC()
for {
now := time.Now().UTC()
if now.YearDay() != currentTime.YearDay() {
wipeLogFile()
}
if targetFile == nil {
currentFile := head + time.Now().UTC().Format(".2006-01-02.150405") + ext
targetFile, _ = os.OpenFile(currentFile, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
currentTime = time.Now().UTC()
currentFilePath = head + currentTime.Format("2006-01-02.150405") + ext
targetFile, _ = os.OpenFile(currentFilePath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
}
buff, errRead := reader.ReadBytes('\n')
@ -456,11 +549,7 @@ func (hc *houstonClient) launch(meta *procmeta) error {
total += len(buff)
if total > 1024*1024 {
total = 0
targetFile.Close()
targetFile = nil
hc.uploadProcFiles(meta, "logs/*"+ext, true)
wipeLogFile()
}
}
}
@ -517,7 +606,7 @@ func (hc *houstonClient) launch(meta *procmeta) error {
if len(evalfile) > 0 {
evalfile = path.Join(logfolder, evalfile)
} else {
evalfile = path.Join(logfolder, path.Base(meta.cmd.Args[0]))
evalfile = logfolder + "/"
}
go stdReader(meta.name, stdout, index, evalfile+".log")
@ -636,27 +725,17 @@ func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest,
func (hc *houstonClient) uploadProcFiles(child *procmeta, filter string, deleteAfterUpload bool) {
logger.Println("uploadFiles found :", child.version, child.name)
root, matches := findMatchFiles(hc.config.StorageRoot, child.name, child.version, filter)
go func(deleteAfterUpload bool, root string, matches []string) {
zipFile, err := zipCompressFiles(root, matches)
if err == nil && len(zipFile) > 0 {
if err = hc.uploadZipLogFile(zipFile, child.name, child.version); err == nil {
_, matches := findMatchFiles(hc.config.StorageRoot, child.name, child.version, filter)
go func() {
for _, filename := range matches {
if err := hc.uploadRawLogFile(filename, child.name, child.version); err != nil {
break
}
if deleteAfterUpload {
for _, fn := range matches {
if len(fn) > 0 {
os.Remove(fn)
os.Remove(filename)
}
}
}
} else {
logger.Println("uploadZipLogFile failed :", err)
}
os.Remove(zipFile)
} else if err != nil {
logger.Println("zipLogFiles failed :", err)
}
}(deleteAfterUpload, root, matches)
}()
}
func (hc *houstonClient) uploadFiles(req *shared.UploadRequest) error {
@ -670,22 +749,15 @@ func (hc *houstonClient) uploadFiles(req *shared.UploadRequest) error {
// 실행 중이 아닌 폴더에서도 대상을 찾는다
// 전체 파일을 대상으로
root, matches := findMatchFiles(hc.config.StorageRoot, req.Name, req.Version, req.Filter)
zipFile, err := zipCompressFiles(root, matches)
if err == nil && len(zipFile) > 0 && len(zipFile) > 0 {
if err = hc.uploadZipLogFile(zipFile, req.Name, req.Version); err == nil {
for _, fn := range matches {
if len(fn) > 0 {
os.Remove(fn)
_, matches := findMatchFiles(hc.config.StorageRoot, req.Name, req.Version, req.Filter)
go func() {
for _, filename := range matches {
if err := hc.uploadRawLogFile(filename, req.Name, req.Version); err != nil {
break
}
os.Remove(filename)
}
os.Remove(zipFile)
} else {
logger.Println("uploadZipLogFile failed :", err)
}
} else if err != nil {
logger.Println("zipLogFiles failed :", err)
}
}()
return nil
}

1
go.mod
View File

@ -4,6 +4,7 @@ go 1.19
require (
github.com/Knetic/govaluate v3.0.0+incompatible
github.com/djherbis/times v1.6.0
github.com/go-kit/log v0.2.1
github.com/prometheus/client_golang v1.17.0
github.com/prometheus/common v0.44.0

3
go.sum
View File

@ -22,6 +22,8 @@ github.com/dennwc/ioctl v1.0.0 h1:DsWAAjIxRqNcLn9x6mwfuf2pet3iB7aK90K4tF16rLg=
github.com/dennwc/ioctl v1.0.0/go.mod h1:ellh2YB5ldny99SBU/VX7Nq0xiZbHphf1DrtHxxjMk0=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/djherbis/times v1.6.0 h1:w2ctJ92J8fBvWPxugmXIv7Nz7Q3iDMKNx9v5ocVH20c=
github.com/djherbis/times v1.6.0/go.mod h1:gOHeRAz2h+VJNZ5Gmc/o7iD9k4wW7NMVqieYCY99oc0=
github.com/ema/qdisc v0.0.0-20230120214811-5b708f463de3 h1:Jrl8sD8wO34+EE1dV2vhOXrqFAZa/FILDnZRaV28+cw=
github.com/ema/qdisc v0.0.0-20230120214811-5b708f463de3/go.mod h1:FhIc0fLYi7f+lK5maMsesDqwYojIOh3VfRs8EVd5YJQ=
github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
@ -146,6 +148,7 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211031064116-611d5d643895/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

19
server/http_handler.go
View File

@ -1,7 +1,6 @@
package server
import (
"archive/zip"
"crypto/md5"
"encoding/hex"
"fmt"
@ -141,25 +140,13 @@ func (h *houstonHandler) RegisterHandlers(serveMux gocommon.ServerMuxInterface,
dir := path.Join(h.downloadPath, name, version)
if err := os.MkdirAll(dir, 0775); err == nil {
filepath := path.Join(dir, filename)
localfile, _ := os.Create(filepath)
// filepath가 이미 있으면 append
localfile, _ := os.OpenFile(filepath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0666)
logger.Println("file uploaded :", localfile)
if localfile != nil {
defer localfile.Close()
if _, err = io.Copy(localfile, r.Body); err != nil {
w.WriteHeader(http.StatusInternalServerError)
} else {
localfile.Close()
localfile, _ = os.Open(filepath)
if strings.HasSuffix(filename, ".zip") {
stat, _ := localfile.Stat()
zipreader, _ := zip.NewReader(localfile, stat.Size())
for _, f := range zipreader.File {
file, _ := os.Create(path.Join(dir, f.Name))
comp, _ := f.Open()
io.Copy(file, comp)
file.Close()
}
os.Remove(filepath)
}
}
} else {
w.WriteHeader(http.StatusInternalServerError)