ziggi/minilb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working commit

Browse Source
This commit is contained in:
Олег Бородин
2026-03-24 23:18:34 +02:00
parent 8efe7090be
commit 1f5b4a71f1
28 changed files with 1623 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config/variant.go
+4 -4
View File
@@ -1,9 +1,9 @@
package config
const (
confdirPath = "/home/ziggi/Projects/gserver/etc/minilb"
rundirPath = "/home/ziggi/Projects/gserver/tmp/run"
logdirPath = "/home/ziggi/Projects/gserver/tmp/log"
datadirPath = "/home/ziggi/Projects/gserver/tmp/data"
confdirPath = "/home/ziggi/Projects/minilb/etc/minilb"
rundirPath = "/home/ziggi/Projects/minilb/tmp/run"
logdirPath = "/home/ziggi/Projects/minilb/tmp/log"
datadirPath = "/home/ziggi/Projects/minilb/tmp/data"
packageVersion = "0.0.1"
)
app/handler/handler.go
+3 -3
View File
@@ -15,13 +15,13 @@ type HandlerConfig struct {
type Handler struct {
mlbctl.UnimplementedControlServer
lg *operator.Operator
log *logger.Logger
oper *operator.Operator
log *logger.Logger
}
func NewHandler(conf *HandlerConfig) *Handler {
hand := Handler{
lg: conf.Operator,
oper: conf.Operator,
}
hand.log = logger.NewLogger("handler")
return &hand
app/handler/hello.go
+1 -1
View File
@@ -9,6 +9,6 @@ import (
func (hand *Handler) GetHello(ctx context.Context, req *mlbctl.GetHelloParams) (*mlbctl.GetHelloResult, error) {
var err error
hand.log.Debugf("Handle getHello request")
res, err := hand.lg.GetHello(ctx, req)
res, err := hand.oper.GetHello(ctx, req)
return res, err
}
app/handler/proxy.go
+25
View File
@@ -0,0 +1,25 @@
package handler
import (
"context"
"helmet/pkg/mlbctl"
)
func (hand *Handler) ListForwarders(ctx context.Context, req *mlbctl.ListForwardersParams) (*mlbctl.ListForwardersResult, error) {
var err error
res, err := hand.oper.ListForwarders(ctx, req)
return res, err
}
func (hand *Handler) CreateForwarder(ctx context.Context, req *mlbctl.CreateForwarderParams) (*mlbctl.CreateForwarderResult, error) {
var err error
res, err := hand.oper.CreateForwarder(ctx, req)
return res, err
}
func (hand *Handler) DeleteForwarder(ctx context.Context, req *mlbctl.DeleteForwarderParams) (*mlbctl.DeleteForwarderResult, error) {
var err error
res, err := hand.oper.DeleteForwarder(ctx, req)
return res, err
}
app/operator/oper.go
+6 -3
View File
@@ -4,6 +4,7 @@ import (
"helmet/app/logger"
"helmet/app/config"
"helmet/app/rproxy"
)
type OperatorConfig struct {
@@ -13,13 +14,15 @@ type OperatorConfig struct {
type Operator struct {
log *logger.Logger
auths []config.Auth
proxy *rproxy.Proxy
}
func NewOperator(conf *OperatorConfig) (*Operator, error) {
var err error
lg := &Operator{
oper := &Operator{
auths: conf.Auths,
}
lg.log = logger.NewLogger("operator")
return lg, err
oper.log = logger.NewLogger("operator")
oper.proxy = rproxy.NewProxy()
return oper, err
}
app/operator/proxy.go
+44
View File
@@ -0,0 +1,44 @@
package operator
import (
"context"
"helmet/pkg/mlbctl"
)
func (oper *Operator) ListForwarders(ctx context.Context, params *mlbctl.ListForwardersParams) (*mlbctl.ListForwardersResult, error) {
var err error
res := &mlbctl.ListForwardersResult{
Forwarders: make([]*mlbctl.Forwarder, 0),
}
for _, forw := range oper.proxy.Forwarders {
oForw := &mlbctl.Forwarder{
Lport: forw.Lport,
Dport: forw.Dport,
Destinations: make([]*mlbctl.Destination, 0),
}
for _, dest := range forw.Dests {
oDest := &mlbctl.Destination{
Address: dest.Address,
}
oForw.Destinations = append(oForw.Destinations, oDest)
}
res.Forwarders = append(res.Forwarders, oForw)
}
return res, err
}
func (oper *Operator) CreateForwarder(ctx context.Context, params *mlbctl.CreateForwarderParams) (*mlbctl.CreateForwarderResult, error) {
var err error
res := &mlbctl.CreateForwarderResult{}
err = oper.proxy.AddForwarder(ctx, params.Lport, params.Dport, params.Destinations...)
return res, err
}
func (oper *Operator) DeleteForwarder(ctx context.Context, params *mlbctl.DeleteForwarderParams) (*mlbctl.DeleteForwarderResult, error) {
var err error
res := &mlbctl.DeleteForwarderResult{}
err = oper.proxy.DeleteForwarder(ctx, params.Lport)
return res, err
}
app/rproxy/proxy.go
+192
View File
@@ -0,0 +1,192 @@
package rproxy
import (
"context"
"errors"
"io"
"log"
"math/rand"
"net"
"strconv"
"sync"
)
type Proxy struct {
Forwarders []*Forwarder `json:"forwarders" yaml:"forwarders"`
ctx context.Context `json:"-" yaml:"-"`
cancel context.CancelFunc `json:"-" yaml:"-"`
wg sync.WaitGroup `json:"-" yaml:"-"`
}
func NewProxy() *Proxy {
ctx, cancel := context.WithCancel(context.Background())
return &Proxy{
Forwarders: make([]*Forwarder, 0),
ctx: ctx,
cancel: cancel,
}
}
func (bal *Proxy) AddForwarder(ctx context.Context, lport, dport uint32, addrs ...string) error {
var err error
forw, err := NewForwarder(ctx, lport, dport, addrs...)
if err != nil {
return err
}
bal.Forwarders = append(bal.Forwarders, forw)
bal.wg.Add(1)
go forw.Listen(&bal.wg)
return err
}
func (bal *Proxy) DeleteForwarder(ctx context.Context, lport uint32) error {
var err error
forwarders := make([]*Forwarder, 0)
for _, forw := range bal.Forwarders {
if forw.Lport == lport {
forw.Stop()
continue
}
forwarders = append(forwarders, forw)
}
bal.Forwarders = forwarders
return err
}
func (bal *Proxy) Start() error {
var err error
for _, forw := range bal.Forwarders {
bal.wg.Add(1)
go forw.Listen(&bal.wg)
}
bal.wg.Wait()
return err
}
func (bal *Proxy) Stop() error {
var err error
for _, forw := range bal.Forwarders {
forw.Stop()
}
return err
}
type Forwarder struct {
listen net.Listener `json:"-" yaml:"-"`
ctx context.Context `json:"-" yaml:"-"`
cancel context.CancelFunc `json:"-" yaml:"-"`
Lport uint32 `json:"lport" yaml:"lport"`
Dport uint32 `json:"dport" yaml:"dport"`
Dests []*Destination `json:"dests" yaml:"dests"`
}
func NewForwarder(ctx context.Context, lport, dport uint32, addrs ...string) (*Forwarder, error) {
ctx, cancel := context.WithCancel(ctx)
forw := &Forwarder{
Dests: make([]*Destination, 0),
Lport: lport,
Dport: dport,
ctx: ctx,
cancel: cancel,
}
for _, addr := range addrs {
dest := NewDestination(addr)
forw.Dests = append(forw.Dests, dest)
}
port := ":" + strconv.FormatUint(uint64(forw.Lport), 10)
listen, err := net.Listen("tcp", port)
if err != nil {
return forw, err
}
forw.listen = listen
return forw, err
}
func (forw *Forwarder) Listen(wg *sync.WaitGroup) {
log.Printf("Start listening on %d\n", forw.Lport)
defer wg.Done()
for {
conn, err := forw.listen.Accept()
if err != nil {
log.Printf("Listen err: %v\n", err)
return
}
go forw.handle(forw.ctx, conn)
}
}
func (forw *Forwarder) Stop() error {
return forw.listen.Close()
}
func (forw *Forwarder) handle(ctx context.Context, inconn net.Conn) {
log.Printf("Handler on %d started\n", forw.Lport)
defer inconn.Close()
if len(forw.Dests) == 0 {
return
}
addrnum := rand.Uint32() % uint32(len(forw.Dests))
ipaddr := forw.Dests[addrnum].Address
dstaddr := ipaddr + ":" + strconv.FormatUint(uint64(forw.Dport), 10)
outconn, err := net.Dial("tcp", dstaddr)
if err != nil {
return
}
var wg sync.WaitGroup
wg.Add(1)
go forw.stream(&wg, inconn, outconn)
wg.Add(1)
go forw.stream(&wg, outconn, inconn)
wg.Wait()
log.Printf("Handler on %d stopped\n", forw.Lport)
}
func (forw *Forwarder) stream(wg *sync.WaitGroup, inconn io.Reader, outconn io.Writer) {
defer wg.Done()
_, err := copy(forw.ctx, outconn, inconn)
if err != nil {
log.Printf("Copy err: %v\n", err)
}
}
type Destination struct {
Address string `json:"address" yaml:"address"`
}
func NewDestination(addr string) *Destination {
return &Destination{
Address: addr,
}
}
func copy(ctx context.Context, writer io.Writer, reader io.Reader) (int64, error) {
var err error
var size int64
var halt bool
buffer := make([]byte, 1024*4)
for {
select {
case <-ctx.Done():
err = errors.New("Break copy by context")
break
default:
}
rsize, err := reader.Read(buffer)
if err == io.EOF {
err = nil
halt = true
}
if err != nil {
return size, err
}
wsize, err := writer.Write(buffer[0:rsize])
size += int64(wsize)
if err != nil {
return size, err
}
if halt {
break
}
}
return size, err
}
app/server/server.go
+21 -17
View File
@@ -253,8 +253,10 @@ func (srv *Server) Daemonize() error {
func (srv *Server) Rotator() {
srv.wg.Add(1)
var counter uint64
logFunc := func() {
for {
counter += 1
select {
case <-srv.ctx.Done():
srv.wg.Done()
@@ -262,25 +264,27 @@ func (srv *Server) Rotator() {
return
default:
}
stat, err := srv.logf.Stat()
if err == nil && stat.Size() > srv.conf.LogLimit {
srv.log.Infof("Rotate log file")
countFiles := 3
for i := 1; i < countFiles; i++ {
nextName := fmt.Sprintf("%s.%d", srv.conf.LogPath, i+1)
prevName := fmt.Sprintf("%s.%d", srv.conf.LogPath, i)
os.Rename(prevName, nextName)
}
os.Rename(srv.conf.LogPath, srv.conf.LogPath+".1")
logFile, err := os.OpenFile(srv.conf.LogPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0640)
if err == nil {
syscall.Dup2(int(logFile.Fd()), int(os.Stdout.Fd()))
syscall.Dup2(int(logFile.Fd()), int(os.Stderr.Fd()))
srv.logf.Close()
srv.logf = logFile
if (counter % 60) == 1 {
stat, err := srv.logf.Stat()
if err == nil && stat.Size() > srv.conf.LogLimit {
srv.log.Infof("Rotate log file")
countFiles := 3
for i := 1; i < countFiles; i++ {
nextName := fmt.Sprintf("%s.%d", srv.conf.LogPath, i+1)
prevName := fmt.Sprintf("%s.%d", srv.conf.LogPath, i)
os.Rename(prevName, nextName)
}
os.Rename(srv.conf.LogPath, srv.conf.LogPath+".1")
logFile, err := os.OpenFile(srv.conf.LogPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0640)
if err == nil {
syscall.Dup2(int(logFile.Fd()), int(os.Stdout.Fd()))
syscall.Dup2(int(logFile.Fd()), int(os.Stderr.Fd()))
srv.logf.Close()
srv.logf = logFile
}
}
}
time.Sleep(3 * time.Second)
time.Sleep(1 * time.Second)
}
}
go logFunc()