refactor: folder structure for proxies

pkg/filters/meshadaptor/meshadaptor.go

@@ -20,7 +20,7 @@ package meshadaptor
 import (
  "github.com/megaease/easegress/pkg/context"
  "github.com/megaease/easegress/pkg/filters"
- "github.com/megaease/easegress/pkg/filters/proxy"
+ proxy "github.com/megaease/easegress/pkg/filters/proxies/httpproxy"
  "github.com/megaease/easegress/pkg/protocols/httpprot"
  "github.com/megaease/easegress/pkg/protocols/httpprot/httpheader"
  "github.com/megaease/easegress/pkg/util/pathadaptor"

pkg/filters/proxies/basepool.go

@@ -0,0 +1,163 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * 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
+ *
+ * https://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 proxies
+
+/*
+// BaseServerPool defines a server pool.
+type BaseServerPool struct {
+ name string
+ done chan struct{}
+ wg sync.WaitGroup
+ loadBalancer atomic.Value
+}
+
+// BaseServerPoolSpec is the spec for a base server pool.
+type BaseServerPoolSpec struct {
+ ServerTags []string `json:"serverTags" jsonschema:"omitempty,uniqueItems=true"`
+ Servers []*Server `json:"servers" jsonschema:"omitempty"`
+ ServiceRegistry string `json:"serviceRegistry" jsonschema:"omitempty"`
+ ServiceName string `json:"serviceName" jsonschema:"omitempty"`
+ LoadBalance *LoadBalanceSpec `json:"loadBalance" jsonschema:"omitempty"`
+}
+
+// Validate validates ServerPoolSpec.
+func (sps *BaseServerPoolSpec) Validate() error {
+ if sps.ServiceName == "" && len(sps.Servers) == 0 {
+ return fmt.Errorf("both serviceName and servers are empty")
+ }
+
+ serversGotWeight := 0
+ for _, server := range sps.Servers {
+ if server.Weight > 0 {
+ serversGotWeight++
+ }
+ }
+ if serversGotWeight > 0 && serversGotWeight < len(sps.Servers) {
+ msgFmt := "not all servers have weight(%d/%d)"
+ return fmt.Errorf(msgFmt, serversGotWeight, len(sps.Servers))
+ }
+
+ if sps.ServiceName != "" && sps.LoadBalance.HealthCheck != nil {
+ return fmt.Errorf("can not open health check for service discovery")
+ }
+
+ return nil
+}
+
+// Init initialize the base server pool according to the spec.
+func (bsp *BaseServerPool) Init(super *supervisor.Supervisor, name string, spec *BaseServerPoolSpec) {
+ bsp.name = name
+ bsp.done = make(chan struct{})
+
+ if spec.ServiceRegistry == "" || spec.ServiceName == "" {
+ bsp.createLoadBalancer(spec.LoadBalance, spec.Servers)
+ return
+ }
+
+ // watch service registry
+ entity := super.MustGetSystemController(serviceregistry.Kind)
+ registry := entity.Instance().(*serviceregistry.ServiceRegistry)
+
+ instances, err := registry.ListServiceInstances(spec.ServiceRegistry, spec.ServiceName)
+ if err != nil {
+ msgFmt := "first try to use service %s/%s failed(will try again): %v"
+ logger.Warnf(msgFmt, spec.ServiceRegistry, spec.ServiceName, err)
+ bsp.createLoadBalancer(spec.LoadBalance, spec.Servers)
+ }
+
+ bsp.useService(spec, instances)
+
+ watcher := registry.NewServiceWatcher(spec.ServiceRegistry, spec.ServiceName)
+ bsp.wg.Add(1)
+ go func() {
+ for {
+ select {
+ case <-bsp.done:
+ watcher.Stop()
+ bsp.wg.Done()
+ return
+ case event := <-watcher.Watch():
+ bsp.useService(spec, event.Instances)
+ }
+ }
+ }()
+}
+
+// LoadBalancer returns the load balancer of the server pool.
+func (bsp *BaseServerPool) LoadBalancer() LoadBalancer {
+ if v := bsp.loadBalancer.Load(); v != nil {
+ return v.(LoadBalancer)
+ }
+ return nil
+}
+
+func (bsp *BaseServerPool) createLoadBalancer(spec *LoadBalanceSpec, servers []*Server) {
+ for _, server := range servers {
+ server.CheckAddrPattern()
+ }
+
+ if spec == nil {
+ spec = &LoadBalanceSpec{}
+ }
+
+ lb := NewLoadBalancer(spec, servers)
+ if old := bsp.loadBalancer.Swap(lb); old != nil {
+ old.(LoadBalancer).Close()
+ }
+}
+
+func (bsp *BaseServerPool) useService(spec *BaseServerPoolSpec, instances map[string]*serviceregistry.ServiceInstanceSpec) {
+ servers := make([]*Server, 0)
+
+ for _, instance := range instances {
+ // default to true in case of sp.spec.ServerTags is empty
+ match := true
+
+ for _, tag := range spec.ServerTags {
+ if match = stringtool.StrInSlice(tag, instance.Tags); match {
+ break
+ }
+ }
+
+ if match {
+ servers = append(servers, &Server{
+ URL: instance.URL(),
+ Tags: instance.Tags,
+ Weight: instance.Weight,
+ })
+ }
+ }
+
+ if len(servers) == 0 {
+ msgFmt := "%s/%s: no service instance satisfy tags: %v"
+ logger.Warnf(msgFmt, spec.ServiceRegistry, spec.ServiceName, spec.ServerTags)
+ servers = spec.Servers
+ }
+
+ bsp.createLoadBalancer(spec.LoadBalance, servers)
+}
+
+func (bsp *BaseServerPool) close() {
+ close(bsp.done)
+ bsp.wg.Wait()
+ if lb := bsp.LoadBalancer(); lb != nil {
+ lb.Close()
+ }
+}
+
+*/

pkg/filters/grpcproxy/codec.go → pkg/filters/proxies/grpcproxy/codec.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "google.golang.org/protobuf/proto"

pkg/filters/grpcproxy/loadbalance.go → pkg/filters/proxies/grpcproxy/loadbalance.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "fmt"

pkg/filters/grpcproxy/loadbalance_test.go → pkg/filters/proxies/grpcproxy/loadbalance_test.go

@@ -15,14 +15,15 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "context"
+ "testing"
+
  "github.com/megaease/easegress/pkg/protocols/grpcprot"
  "github.com/stretchr/testify/assert"
  "google.golang.org/grpc/metadata"
- "testing"
 )
 
 func TestForwardLB(t *testing.T) {

pkg/filters/grpcproxy/pool.go → pkg/filters/proxies/grpcproxy/pool.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  stdcontext "context"

pkg/filters/grpcproxy/proxy.go → pkg/filters/proxies/grpcproxy/proxy.go

@@ -15,14 +15,14 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "fmt"
 
  "github.com/megaease/easegress/pkg/context"
  "github.com/megaease/easegress/pkg/filters"
- "github.com/megaease/easegress/pkg/protocols"
+ "github.com/megaease/easegress/pkg/filters/proxies"
  "github.com/megaease/easegress/pkg/protocols/grpcprot"
  "github.com/megaease/easegress/pkg/resilience"
  "github.com/megaease/easegress/pkg/supervisor"
@@ -84,7 +84,7 @@ type (
  }
 
  // Server is the backend server.
- Server = protocols.Server
+ Server = proxies.Server
 )
 
 // Validate validates Spec.

pkg/filters/grpcproxy/proxy_test.go → pkg/filters/proxies/grpcproxy/proxy_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "os"

pkg/filters/grpcproxy/requestmatch.go → pkg/filters/proxies/grpcproxy/requestmatch.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "fmt"

pkg/filters/grpcproxy/requestmatch_test.go → pkg/filters/proxies/grpcproxy/requestmatch_test.go

@@ -15,17 +15,18 @@
  * limitations under the License.
  */
 
-package grpcprxoy
+package grpcproxy
 
 import (
  "context"
  "fmt"
- "github.com/megaease/easegress/pkg/protocols/grpcprot"
- "google.golang.org/grpc/metadata"
  "math/rand"
  "strconv"
  "testing"
 
+ "github.com/megaease/easegress/pkg/protocols/grpcprot"
+ "google.golang.org/grpc/metadata"
+
  "github.com/stretchr/testify/assert"
 )
 

pkg/filters/proxy/basepool.go → pkg/filters/proxies/httpproxy/basepool.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "fmt"
@@ -33,18 +33,16 @@ type BaseServerPool struct {
  name string
  done chan struct{}
  wg sync.WaitGroup
- filter RequestMatcher
  loadBalancer atomic.Value
 }
 
 // BaseServerPoolSpec is the spec for a base server pool.
 type BaseServerPoolSpec struct {
- Filter *RequestMatcherSpec `json:"filter" jsonschema:"omitempty"`
- ServerTags []string `json:"serverTags" jsonschema:"omitempty,uniqueItems=true"`
- Servers []*Server `json:"servers" jsonschema:"omitempty"`
- ServiceRegistry string `json:"serviceRegistry" jsonschema:"omitempty"`
- ServiceName string `json:"serviceName" jsonschema:"omitempty"`
- LoadBalance *LoadBalanceSpec `json:"loadBalance" jsonschema:"omitempty"`
+ ServerTags []string `json:"serverTags" jsonschema:"omitempty,uniqueItems=true"`
+ Servers []*Server `json:"servers" jsonschema:"omitempty"`
+ ServiceRegistry string `json:"serviceRegistry" jsonschema:"omitempty"`
+ ServiceName string `json:"serviceName" jsonschema:"omitempty"`
+ LoadBalance *LoadBalanceSpec `json:"loadBalance" jsonschema:"omitempty"`
 }
 
 // Validate validates ServerPoolSpec.
@@ -76,10 +74,6 @@ func (bsp *BaseServerPool) Init(super *supervisor.Supervisor, name string, spec
  bsp.name = name
  bsp.done = make(chan struct{})
 
- if spec.Filter != nil {
- bsp.filter = NewRequestMatcher(spec.Filter)
- }
-
  if spec.ServiceRegistry == "" || spec.ServiceName == "" {
  bsp.createLoadBalancer(spec.LoadBalance, spec.Servers)
  return

pkg/filters/proxy/compression.go → pkg/filters/proxies/httpproxy/compression.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "net/http"

pkg/filters/proxy/compression_test.go → pkg/filters/proxies/httpproxy/compression_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "io"

pkg/filters/proxy/loadbalance.go → pkg/filters/proxies/httpproxy/loadbalance.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "crypto/hmac"

pkg/filters/proxy/loadbalance_test.go → pkg/filters/proxies/httpproxy/loadbalance_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "fmt"

pkg/filters/proxy/memorycache.go → pkg/filters/proxies/httpproxy/memorycache.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "net/http"

pkg/filters/proxy/memorycache_test.go → pkg/filters/proxies/httpproxy/memorycache_test.go

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package proxy
+package httpproxy
 
 import (
  "net/http"